//  ---------------------------------------------------------------------
//  Reactor event handlers...
//  Publish our state to peer
960997
func (bstar *Bstar) send_state() (err error) {
	_, err = bstar.statepub.SendMessage(int(bstar.state))
	return
}
885756
func (r *sentReq) stateRequesting() reqStateFn {
	select {
	case ev := <-r.eventsCh:
		r.update(ev)
		switch ev.event {
		case rpSent:
			if ev.peer == nil {
				// request send failed, no more suitable peers
				if r.waiting() {
					// we are already waiting for sent requests which may succeed so keep waiting
					return r.stateNoMorePeers
				}
				// nothing to wait for, no more peers to ask, return with error
				r.stop(light.ErrNoPeers)
				// no need to go to stopped state because waiting() already returned false
				return nil
			}
		case rpSoftTimeout:
			// last request timed out, try asking a new peer
			go r.tryRequest()
			r.lastReqQueued = true
			return r.stateRequesting
		case rpDeliveredInvalid:
			// if it was the last sent request (set to nil by update) then start a new one
			if !r.lastReqQueued && r.lastReqSentTo == nil {
				go r.tryRequest()
				r.lastReqQueued = true
			}
			return r.stateRequesting
		case rpDeliveredValid:
			r.stop(nil)
			return r.stateStopped
		}
		return r.stateRequesting
	case <-r.stopCh:
		return r.stateStopped
	}
}
190568
func (pcc *PubControlClient) Publish(channel string, item *Item) error {
	return pcc.publish(pcc, channel, item)
}
431427
func (c *client) Publish(topic string, qos byte, retained bool, payload interface{}) Token {
	token := newToken(packets.Publish).(*PublishToken)
	DEBUG.Println(CLI, "enter Publish")
	switch {
	case !c.IsConnected():
		token.setError(ErrNotConnected)
		return token
	case c.connectionStatus() == reconnecting && qos == 0:
		token.flowComplete()
		return token
	}
	pub := packets.NewControlPacket(packets.Publish).(*packets.PublishPacket)
	pub.Qos = qos
	pub.TopicName = topic
	pub.Retain = retained
	switch payload.(type) {
	case string:
		pub.Payload = []byte(payload.(string))
	case []byte:
		pub.Payload = payload.([]byte)
	default:
		token.setError(fmt.Errorf("Unknown payload type"))
		return token
	}

	if pub.Qos != 0 && pub.MessageID == 0 {
		pub.MessageID = c.getID(token)
		token.messageID = pub.MessageID
	}
	persistOutbound(c.persist, pub)
	if c.connectionStatus() == reconnecting {
		DEBUG.Println(CLI, "storing publish message (reconnecting), topic:", topic)
	} else {
		DEBUG.Println(CLI, "sending publish message, topic:", topic)
		c.obound <- &PacketAndToken{p: pub, t: token}
	}
	return token
}
238494
func (sw *Switch) addPeer(p Peer) error {
	if err := sw.filterPeer(p); err != nil {
		return err
	}

	p.SetLogger(sw.Logger.With("peer", p.SocketAddr()))

	// Handle the shut down case where the switch has stopped but we're
	// concurrently trying to add a peer.
	if !sw.IsRunning() {
		// XXX should this return an error or just log and terminate?
		sw.Logger.Error("Won't start a peer - switch is not running", "peer", p)
		return nil
	}

	// Start the peer's send/recv routines.
	// Must start it before adding it to the peer set
	// to prevent Start and Stop from being called concurrently.
	err := p.Start()
	if err != nil {
		// Should never happen
		sw.Logger.Error("Error starting peer", "err", err, "peer", p)
		return err
	}

	// Add the peer to PeerSet. Do this before starting the reactors
	// so that if Receive errors, we will find the peer and remove it.
	// Add should not err since we already checked peers.Has().
	if err := sw.peers.Add(p); err != nil {
		return err
	}
	sw.metrics.Peers.Add(float64(1))

	// Start all the reactor protocols on the peer.
	for _, reactor := range sw.reactors {
		reactor.AddPeer(p)
	}

	sw.Logger.Info("Added peer", "peer", p)

	return nil
}
411866
func (nc *Conn) PublishRequest(subj, reply string, data []byte) error {
	return nc.publish(subj, reply, data)
}
1229959
func (n *Nats) PublishRequest(subject string, reply string, data PayloadWithContext) error {
	n.updateContext(data, RequestType_PUBREQ, nil)
	return n.EncCon.PublishRequest(subject, reply, data)
}
1284815
func (s *peerRESTServer) SendEventHandler(w http.ResponseWriter, r *http.Request) {
	if !s.IsValid(w, r) {
		s.writeErrorResponse(w, errors.New("Invalid request"))
		return
	}

	ctx := newContext(r, w, "SendEvent")

	vars := mux.Vars(r)
	bucketName := vars[peerRESTBucket]
	if bucketName == "" {
		s.writeErrorResponse(w, errors.New("Bucket name is missing"))
		return
	}
	var eventReq sendEventRequest
	if r.ContentLength <= 0 {
		s.writeErrorResponse(w, errInvalidArgument)
		return
	}

	err := gob.NewDecoder(r.Body).Decode(&eventReq)
	if err != nil {
		s.writeErrorResponse(w, err)
		return
	}

	var eventResp sendEventResp
	eventResp.Success = true
	errs := globalNotificationSys.send(bucketName, eventReq.Event, eventReq.TargetID)

	for i := range errs {
		reqInfo := (&logger.ReqInfo{}).AppendTags("Event", eventReq.Event.EventName.String())
		reqInfo.AppendTags("targetName", eventReq.TargetID.Name)
		ctx := logger.SetReqInfo(context.Background(), reqInfo)
		logger.LogIf(ctx, errs[i].Err)

		eventResp.Success = false
		s.writeErrorResponse(w, errs[i].Err)
		return
	}
	logger.LogIf(ctx, gob.NewEncoder(w).Encode(&eventResp))
	w.(http.Flusher).Flush()
}
1089228
func (p *Publisher) Publish(v interface{}) {
	p.m.RLock()
	if len(p.subscribers) == 0 {
		p.m.RUnlock()
		return
	}

	wg := wgPool.Get().(*sync.WaitGroup)
	for sub, topic := range p.subscribers {
		wg.Add(1)
		go p.sendTopic(sub, topic, v, wg)
	}
	wg.Wait()
	wgPool.Put(wg)
	p.m.RUnlock()
}
652462
func (srv *Instance) PublishEvent(e lifecycle.Event) error {
	return lifecycle.PublishEvent(e, srv.connector)
}
199096
func (h *apiExecutor) Publish(ctx context.Context, cmd *PublishRequest) *PublishResponse {
	defer func(started time.Time) {
		apiCommandDurationSummary.WithLabelValues(h.protocol, "publish").Observe(time.Since(started).Seconds())
	}(time.Now())

	ch := cmd.Channel
	data := cmd.Data

	resp := &PublishResponse{}

	if ch == "" {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "channel required for publish", nil))
		resp.Error = ErrorBadRequest
		return resp
	}

	if len(data) == 0 {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "data required for publish", nil))
		resp.Error = ErrorBadRequest
		return resp
	}

	_, ok := h.node.ChannelOpts(ch)
	if !ok {
		resp.Error = ErrorNamespaceNotFound
		return resp
	}

	err := h.node.Publish(cmd.Channel, cmd.Data)
	if err != nil {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "error publishing message in engine", map[string]interface{}{"error": err.Error()}))
		resp.Error = ErrorInternal
		return resp
	}
	return resp
}
1228417
func (c *DefaultClient) PublishAppEvent(appID string, eventType types.EventType, payload interface{}) Token {
	topic := ApplicationTopic{appID, AppEvents, string(eventType)}
	msg, err := json.Marshal(payload)
	if err != nil {
		return &simpleToken{fmt.Errorf("Unable to marshal the message payload: %s", err)}
	}
	return c.publish(topic.String(), msg)
}
627179
func (o *Observer) ServeHTTP(rw http.ResponseWriter, r *http.Request) {

	rwf := rw.(http.Flusher)
	closed := rw.(http.CloseNotifier).CloseNotify()

	notify := make(chan []byte)
	o.newObservers <- notify

	defer func() {
		o.closedObservers <- notify
	}()

	rw.Header().Set("Content-Type", "text/event-stream")
	rw.Header().Set("Connection", "keep-alive")
	rwf.Flush()

	for {
		select {
		case evt := <-notify:
			// Write the event to the client. we ignore errors here, and
			// let the close channel tell us when to stop writing.
			rw.Write(evt)
			rwf.Flush()

		case <-closed: // client has disconnected. Exit.
			return
		case <-o.closed: // The Observer is shutting down. Exit.
			return
		}
	}
}
666680
func (b *RTMBroker) Publish(e Publishable) error {
	d, err := e.Publishable()
	if err != nil {
		return err
	}

	return b.conn.WriteMessage(websocket.TextMessage, d)
}
135488
func (p *Publisher) Publish(topic string, body []byte) error {
	pub := p.topic(topic)
	res := pub.Publish(context.Background(), &pubsub.Message{
		Data: body,
	})
	_, err := res.Get(context.Background())
	return err
}
925611
func (s *Statsd) publish(msgs []sendable) {
	for _, buf := range bufferize(msgs, maxPacketSize) {
		// In the base case, when the Statsd struct is backed by a net.Conn,
		// "Multiple goroutines may invoke methods on a Conn simultaneously."
		//   -- http://golang.org/pkg/net/#Conn
		// Otherwise, Bring Your Own Synchronization™.
		n, err := s.w.Write(buf)
		if err != nil {
			log.Printf("g2s: publish: %s", err)
		} else if n != len(buf) {
			log.Printf("g2s: publish: short send: %d < %d", n, len(buf))
		}
	}
}
1173937
func (c *Client) processPublish(publish *packet.Publish) error {
	// handle qos 0 flow
	if publish.Message.QOS == 0 {
		// publish message
		err := c.backend.Publish(c, &publish.Message, nil)
		if err != nil {
			return c.die(BackendError, err)
		}

		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)

		return nil
	}

	// acquire publish token
	select {
	case <-c.publishTokens:
		// continue
	case <-time.After(c.TokenTimeout):
		return c.die(ClientError, ErrTokenTimeout)
	case <-c.tomb.Dying():
		return tomb.ErrDying
	}

	// handle qos 1 flow
	if publish.Message.QOS == 1 {
		// prepare puback
		puback := packet.NewPuback()
		puback.ID = publish.ID

		// publish message and queue puback if ack is called
		err := c.backend.Publish(c, &publish.Message, func() {
			c.backend.Log(MessageAcknowledged, c, nil, &publish.Message, nil)

			select {
			case c.ackQueue <- puback:
			case <-c.tomb.Dying():
			}
		})
		if err != nil {
			return c.die(BackendError, err)
		}

		c.backend.Log(MessagePublished, c, nil, &publish.Message, nil)
	}

	// handle qos 2 flow
	if publish.Message.QOS == 2 {
		// store received publish packet in session
		err := c.session.SavePacket(session.Incoming, publish)
		if err != nil {
			return c.die(SessionError, err)
		}

		// prepare pubrec packet
		pubrec := packet.NewPubrec()
		pubrec.ID = publish.ID

		// signal qos 2 pubrec
		err = c.send(pubrec, true)
		if err != nil {
			return c.die(TransportError, err)
		}
	}

	return nil
}