// IpnsRepublisher runs new IPNS republisher service
979096
func IpnsRepublisher(repubPeriod time.Duration, recordLifetime time.Duration) func(lcProcess, namesys.NameSystem, repo.Repo, crypto.PrivKey) error {
	return func(lc lcProcess, namesys namesys.NameSystem, repo repo.Repo, privKey crypto.PrivKey) error {
		repub := republisher.NewRepublisher(namesys, repo.Datastore(), privKey, repo.Keystore())

		if repubPeriod != 0 {
			if !util.Debug && (repubPeriod < time.Minute || repubPeriod > (time.Hour*24)) {
				return fmt.Errorf("config setting IPNS.RepublishPeriod is not between 1min and 1day: %s", repubPeriod)
			}

			repub.Interval = repubPeriod
		}

		if recordLifetime != 0 {
			repub.RecordLifetime = recordLifetime
		}

		lc.Append(repub.Run)
		return nil
	}
}
120590
func chPublish(c *EncodedConn, chVal reflect.Value, subject string) {
	for {
		val, ok := chVal.Recv()
		if !ok {
			// Channel has most likely been closed.
			return
		}
		if e := c.Publish(subject, val.Interface()); e != nil {
			// Do this under lock.
			c.Conn.mu.Lock()
			defer c.Conn.mu.Unlock()

			if c.Conn.Opts.AsyncErrorCB != nil {
				// FIXME(dlc) - Not sure this is the right thing to do.
				// FIXME(ivan) - If the connection is not yet closed, try to schedule the callback
				if c.Conn.isClosed() {
					go c.Conn.Opts.AsyncErrorCB(c.Conn, nil, e)
				} else {
					c.Conn.ach.push(func() { c.Conn.Opts.AsyncErrorCB(c.Conn, nil, e) })
				}
			}
			return
		}
	}
}
284894
func (cli *NetworkCli) CmdServiceUnpublish(chain string, args ...string) error {
	cmd := cli.Subcmd(chain, "unpublish", "SERVICE[.NETWORK]", "Removes a service", false)
	force := cmd.Bool([]string{"f", "-force"}, false, "force unpublish service")
	cmd.Require(flag.Exact, 1)
	err := cmd.ParseFlags(args, true)
	if err != nil {
		return err
	}

	sn, nn := parseServiceName(cmd.Arg(0))
	serviceID, err := lookupServiceID(cli, nn, sn)
	if err != nil {
		return err
	}

	sd := serviceDelete{Name: sn, Force: *force}
	_, _, err = readBody(cli.call("DELETE", "/services/"+serviceID, sd, nil))

	return err
}
922327
func (s *NetworkService) NewReleasePublicIpRangeParams(id string) *ReleasePublicIpRangeParams {
	p := &ReleasePublicIpRangeParams{}
	p.p = make(map[string]interface{})
	p.p["id"] = id
	return p
}
187476
func (ks *KeyStore) GetPublickeyPeers(keyid string) (topic []Topic, address []PssAddress, err error) {
	ks.mx.RLock()
	defer ks.mx.RUnlock()
	for t, peer := range ks.pubKeyPool[keyid] {
		topic = append(topic, t)
		address = append(address, peer.address)
	}
	return topic, address, nil
}
258185
func (cli *NetworkCli) CmdServicePublish(chain string, args ...string) error {
	cmd := cli.Subcmd(chain, "publish", "SERVICE[.NETWORK]", "Publish a new service on a network", false)
	flAlias := opts.NewListOpts(netutils.ValidateAlias)
	cmd.Var(&flAlias, []string{"-alias"}, "Add alias to self")
	cmd.Require(flag.Exact, 1)
	err := cmd.ParseFlags(args, true)
	if err != nil {
		return err
	}

	sn, nn := parseServiceName(cmd.Arg(0))
	sc := serviceCreate{Name: sn, Network: nn, MyAliases: flAlias.GetAll()}
	obj, _, err := readBody(cli.call("POST", "/services", sc, nil))
	if err != nil {
		return err
	}

	var replyID string
	err = json.Unmarshal(obj, &replyID)
	if err != nil {
		return err
	}

	fmt.Fprintf(cli.out, "%s\n", replyID)
	return nil
}
90236
func NewPublisher(
	publisher *nats.Conn,
	subject string,
	enc EncodeRequestFunc,
	dec DecodeResponseFunc,
	options ...PublisherOption,
) *Publisher {
	p := &Publisher{
		publisher: publisher,
		subject:   subject,
		enc:       enc,
		dec:       dec,
		timeout:   10 * time.Second,
	}
	for _, option := range options {
		option(p)
	}
	return p
}
851447
func NewPublisher(cmInformer coreinformers.ConfigMapInformer, nsInformer coreinformers.NamespaceInformer, cl clientset.Interface, rootCA []byte) (*Publisher, error) {
	e := &Publisher{
		client: cl,
		rootCA: rootCA,
		queue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "root_ca_cert_publisher"),
	}
	if cl.CoreV1().RESTClient().GetRateLimiter() != nil {
		if err := metrics.RegisterMetricAndTrackRateLimiterUsage("root_ca_cert_publisher", cl.CoreV1().RESTClient().GetRateLimiter()); err != nil {
			return nil, err
		}
	}

	cmInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		DeleteFunc: e.configMapDeleted,
		UpdateFunc: e.configMapUpdated,
	})
	e.cmLister = cmInformer.Lister()
	e.cmListerSynced = cmInformer.Informer().HasSynced

	nsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    e.namespaceAdded,
		UpdateFunc: e.namespaceUpdated,
	})
	e.nsListerSynced = nsInformer.Informer().HasSynced

	e.syncHandler = e.syncNamespace

	return e, nil

}
225860
func (az *Cloud) DeletePublicIP(service *v1.Service, pipResourceGroup string, pipName string) error {
	if az.Config.shouldOmitCloudProviderBackoff() {
		ctx, cancel := getContextWithCancel()
		defer cancel()

		resp, err := az.PublicIPAddressesClient.Delete(ctx, pipResourceGroup, pipName)
		return az.processHTTPResponse(service, "DeletePublicIPAddress", resp, err)
	}

	return az.deletePublicIPWithRetry(service, pipResourceGroup, pipName)
}
1155855
func getpublicIPAddressID(subscriptionID string, resourceGroupName, pipName string) string {
	return fmt.Sprintf(
		publicIPAddressIDTemplate,
		subscriptionID,
		resourceGroupName,
		pipName)
}
1139570
func (p *IpnsPublisher) Publish(ctx context.Context, k ci.PrivKey, value path.Path) error {
	log.Debugf("Publish %s", value)
	return p.PublishWithEOL(ctx, k, value, time.Now().Add(DefaultRecordEOL))
}
764671
func (az *Cloud) deletePublicIPWithRetry(service *v1.Service, pipResourceGroup string, pipName string) error {
	return wait.ExponentialBackoff(az.requestBackoff(), func() (bool, error) {
		ctx, cancel := getContextWithCancel()
		defer cancel()

		resp, err := az.PublicIPAddressesClient.Delete(ctx, pipResourceGroup, pipName)
		return az.processHTTPRetryResponse(service, "DeletePublicIPAddress", resp, err)
	})
}
622893
func NewPublisher(topic string, c client.Client) Publisher {
	if c == nil {
		c = client.NewClient()
	}
	return &publisher{c, topic}
}
1093141
func NewPUBLISH(opts *PUBLISHOptions) (Packet, error) {
	// Initialize the options.
	if opts == nil {
		opts = &PUBLISHOptions{}
	}

	// Validate the options.
	if err := opts.validate(); err != nil {
		return nil, err
	}

	// Create a PUBLISH Packet.
	p := &PUBLISH{
		DUP:       opts.DUP,
		QoS:       opts.QoS,
		retain:    opts.Retain,
		TopicName: opts.TopicName,
		PacketID:  opts.PacketID,
		Message:   opts.Message,
	}

	// Set the variable header to the Packet.
	p.setVariableHeader()

	// Set the payload to the Packet.
	p.setPayload()

	// Set the Fixed header to the Packet.
	p.setFixedHeader()

	// Return the Packet.
	return p, nil
}
968119
func (f *NodeClient) NodePublishVolume(ctx context.Context, req *csipb.NodePublishVolumeRequest, opts ...grpc.CallOption) (*csipb.NodePublishVolumeResponse, error) {
	if f.nextErr != nil {
		return nil, f.nextErr
	}

	if req.GetVolumeId() == "" {
		return nil, errors.New("missing volume id")
	}
	if req.GetTargetPath() == "" {
		return nil, errors.New("missing target path")
	}
	fsTypes := "block|ext4|xfs|zfs"
	fsType := req.GetVolumeCapability().GetMount().GetFsType()
	if !strings.Contains(fsTypes, fsType) {
		return nil, errors.New("invalid fstype")
	}
	f.nodePublishedVolumes[req.GetVolumeId()] = CSIVolume{
		VolumeHandle:    req.GetVolumeId(),
		Path:            req.GetTargetPath(),
		DeviceMountPath: req.GetStagingTargetPath(),
		VolumeContext:   req.GetVolumeContext(),
		FSType:          req.GetVolumeCapability().GetMount().GetFsType(),
		MountFlags:      req.GetVolumeCapability().GetMount().MountFlags,
	}
	return &csipb.NodePublishVolumeResponse{}, nil
}
1337207
func (f *Facade) GetAllPublicEndpoints(ctx datastore.Context) ([]service.PublicEndpoint, error) {
	defer ctx.Metrics().Stop(ctx.Metrics().Start("GetAllPublicEndpoints"))
	return f.serviceStore.GetAllPublicEndpoints(ctx)
}
674314
func (w *workManager) AddPublishWorker() {
	nw := newWorker(w.publishchan)
	go nw.start()
	w.publishWkrs = append(w.publishWkrs, nw)
	w.publishWkrSize++
}