// subHandler just spins on the socket and publishes messages.  It sends
// "PUB #<count> <time>".  Not very interesting...
10086
func subHandler(sock mangos.Socket) {
	count := 0
	for {
		msg := fmt.Sprintf("PUB #%d %s", count, time.Now().String())
		if e := sock.Send([]byte(msg)); e != nil {
			die("Cannot send pub: %v", e)
		}
		time.Sleep(5 * time.Second)
		count++
	}
}
509682
func (o Middleware) SubscribeInterceptor(opts pubsub.HandlerOptions, next pubsub.MsgHandler) pubsub.MsgHandler {
	return func(ctx context.Context, m pubsub.Msg) error {
		if o.Logger == nil {
			o.Logger = DefaultLogger()
		}

		start := time.Now()
		err := next(ctx, m)
		elapsed := time.Now().Sub(start)

		var traceID string
		span := opentracing.SpanFromContext(ctx)
		if span != nil {
			zs, ok := span.Context().(zipkintracing.SpanContext)
			if ok {
				traceID = zs.TraceID.ToHex()
			}
		}

		fields := []zapcore.Field{
			zap.String("component", "pubsub"),
			zap.String("topic", opts.Topic),
			zap.String("handler", opts.Name),
			zap.Duration("duration", elapsed),
			zap.Error(err),
		}

		if m.ID != "" {
			fields = append(fields, zap.String("id", m.ID))
		}

		if traceID != "" {
			fields = append(fields, zap.String("trace-id", traceID))
		}

		o.Logger.Debug("Processed PubSub Msg",
			fields...,
		)

		return err
	}
}
967012
func (p *Pubsub) Subscribe(ev EventBuilder, l Listener) {
	timer := gaugeLatency(PromSubLatency)
	defer timer()

	p.subsMu.Lock()
	count := p.subs[ev.kind].Add(ev, l)
	p.subsMu.Unlock()

	if count == 1 {
		PromSubscriptions.Inc()
		written := make(chan struct{}, 1)
		p.send <- command{
			command: ev.kind.SubCommand(),
			channel: ev.Name(),
			written: written,
		}

		<-written
	}
}
75476
func (c *Client) messageHandler() {
	for {
		if c.scanner.Scan() {
			line := c.scanner.Text()
			if line == "error id=0 msg=ok" {
				c.err <- nil
			} else if matches := respTrailerRe.FindStringSubmatch(line); len(matches) == 4 {
				c.err <- NewError(matches)
			} else if strings.Index(line, "notify") == 0 {
				if n, err := decodeNotification(line); err == nil {
					// non-blocking write
					select {
					case c.notify <- n:
					default:
					}
				}
			} else {
				c.res = append(c.res, line)
			}
		} else {
			err := c.scanErr()
			c.err <- err
			if err == io.ErrUnexpectedEOF {
				close(c.disconnect)
				return
			}
		}
	}
}
1236898
func (nc *Conn) Subscribe(subj string, cb MsgHandler) (*Subscription, error) {
	return nc.subscribe(subj, _EMPTY_, cb, nil, false)
}
1136827
func (pm *Manager) SubscribeEvents(buffer int, watchEvents ...Event) (eventCh <-chan interface{}, cancel func()) {
	topic := func(i interface{}) bool {
		observed, ok := i.(Event)
		if !ok {
			panic(fmt.Sprintf("unexpected type passed to event channel: %v", reflect.TypeOf(i)))
		}
		for _, e := range watchEvents {
			if e.matches(observed) {
				return true
			}
		}
		// If no specific events are specified always assume a matched event
		// If some events were specified and none matched above, then the event
		// doesn't match
		return watchEvents == nil
	}
	ch := pm.publisher.SubscribeTopicWithBuffer(topic, buffer)
	cancelFunc := func() { pm.publisher.Evict(ch) }
	return ch, cancelFunc
}
385886
func (cln *Client) Subscribe(c *ble.Characteristic, ind bool, fn ble.NotificationHandler) error {
	cln.conn.Lock()
	defer cln.conn.Unlock()
	cln.conn.subs[c.Handle] = &sub{fn: fn, char: c}
	rsp := cln.conn.sendReq(68, xpc.Dict{
		"kCBMsgArgDeviceUUID":                cln.id,
		"kCBMsgArgCharacteristicHandle":      c.Handle,
		"kCBMsgArgCharacteristicValueHandle": c.ValueHandle,
		"kCBMsgArgState":                     1,
	})
	if rsp.err() != nil {
		delete(cln.conn.subs, c.Handle)
		return rsp.err()
	}
	return nil
}
255643
func MultiHandler(handlers ...Handler) Handler {
	multi := make([]Handler, 0, len(handlers))

	for _, h := range handlers {
		if h != nil {
			if m, ok := h.(*multiHandler); ok {
				multi = append(multi, m.handlers...) // flatten multi handlers
			} else {
				multi = append(multi, h)
			}
		}
	}

	if len(multi) == 1 {
		return multi[0]
	}

	return &multiHandler{handlers: multi}
}
998560
func (p *Publisher) SubChannel(finalMsg interface{}) (msgChan <-chan interface{}, lastMsg interface{}) {
	listener, cur := p.SubReader()
	outch := make(chan interface{})
	go listen(listener, outch, finalMsg)
	return outch, cur
}
592549
func (p *PubSub) Subscribe(id string, s Subscriber) error {
	p.m.Lock()
	defer p.m.Unlock()

	p.subscribers[id] = s
	return nil
}
19347
func (p *peer) readHandler() {
	defer p.wg.Done()

	// We'll stop the timer after a new messages is received, and also
	// reset it after we process the next message.
	idleTimer := time.AfterFunc(idleTimeout, func() {
		err := fmt.Errorf("Peer %s no answer for %s -- disconnecting",
			p, idleTimeout)
		p.Disconnect(err)
	})

	// Initialize our negotiated gossip sync method before reading messages
	// off the wire. When using gossip queries, this ensures a gossip
	// syncer is active by the time query messages arrive.
	//
	// TODO(conner): have peer store gossip syncer directly and bypass
	// gossiper?
	p.initGossipSync()

	discStream := newDiscMsgStream(p)
	discStream.Start()
	defer discStream.Stop()

	chanMsgStreams := make(map[lnwire.ChannelID]*msgStream)
out:
	for atomic.LoadInt32(&p.disconnect) == 0 {
		nextMsg, err := p.readNextMessage()
		if !idleTimer.Stop() {
			select {
			case <-idleTimer.C:
			default:
			}
		}
		if err != nil {
			peerLog.Infof("unable to read message from %v: %v",
				p, err)

			switch err.(type) {
			// If this is just a message we don't yet recognize,
			// we'll continue processing as normal as this allows
			// us to introduce new messages in a forwards
			// compatible manner.
			case *lnwire.UnknownMessage:
				idleTimer.Reset(idleTimeout)
				continue

			// If they sent us an address type that we don't yet
			// know of, then this isn't a dire error, so we'll
			// simply continue parsing the remainder of their
			// messages.
			case *lnwire.ErrUnknownAddrType:
				idleTimer.Reset(idleTimeout)
				continue

			// If the NodeAnnouncement has an invalid alias, then
			// we'll log that error above and continue so we can
			// continue to read messges from the peer.
			case *lnwire.ErrInvalidNodeAlias:
				idleTimer.Reset(idleTimeout)
				continue

			// If the error we encountered wasn't just a message we
			// didn't recognize, then we'll stop all processing s
			// this is a fatal error.
			default:
				break out
			}
		}

		var (
			isChanUpdate bool
			targetChan   lnwire.ChannelID
		)

		switch msg := nextMsg.(type) {
		case *lnwire.Pong:
			// When we receive a Pong message in response to our
			// last ping message, we'll use the time in which we
			// sent the ping message to measure a rough estimate of
			// round trip time.
			pingSendTime := atomic.LoadInt64(&p.pingLastSend)
			delay := (time.Now().UnixNano() - pingSendTime) / 1000
			atomic.StoreInt64(&p.pingTime, delay)

		case *lnwire.Ping:
			pongBytes := make([]byte, msg.NumPongBytes)
			p.queueMsg(lnwire.NewPong(pongBytes), nil)

		case *lnwire.OpenChannel:
			p.server.fundingMgr.processFundingOpen(msg, p)
		case *lnwire.AcceptChannel:
			p.server.fundingMgr.processFundingAccept(msg, p)
		case *lnwire.FundingCreated:
			p.server.fundingMgr.processFundingCreated(msg, p)
		case *lnwire.FundingSigned:
			p.server.fundingMgr.processFundingSigned(msg, p)
		case *lnwire.FundingLocked:
			p.server.fundingMgr.processFundingLocked(msg, p)

		case *lnwire.Shutdown:
			select {
			case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
			case <-p.quit:
				break out
			}
		case *lnwire.ClosingSigned:
			select {
			case p.chanCloseMsgs <- &closeMsg{msg.ChannelID, msg}:
			case <-p.quit:
				break out
			}

		case *lnwire.Error:
			key := p.addr.IdentityKey

			switch {
			// In the case of an all-zero channel ID we want to
			// forward the error to all channels with this peer.
			case msg.ChanID == lnwire.ConnectionWideID:
				for chanID, chanStream := range chanMsgStreams {
					chanStream.AddMsg(nextMsg)

					// Also marked this channel as failed,
					// so we won't try to restart it on
					// reconnect with this peer.
					p.failedChannels[chanID] = struct{}{}
				}

			// If the channel ID for the error message corresponds
			// to a pending channel, then the funding manager will
			// handle the error.
			case p.server.fundingMgr.IsPendingChannel(msg.ChanID, key):
				p.server.fundingMgr.processFundingError(msg, key)

			// If not we hand the error to the channel link for
			// this channel.
			default:
				isChanUpdate = true
				targetChan = msg.ChanID

				// Also marked this channel as failed, so we
				// won't try to restart it on reconnect with
				// this peer.
				p.failedChannels[targetChan] = struct{}{}
			}

		// TODO(roasbeef): create ChanUpdater interface for the below
		case *lnwire.UpdateAddHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFulfillHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFailMalformedHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFailHTLC:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.RevokeAndAck:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.CommitSig:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.UpdateFee:
			isChanUpdate = true
			targetChan = msg.ChanID
		case *lnwire.ChannelReestablish:
			isChanUpdate = true
			targetChan = msg.ChanID

		case *lnwire.ChannelUpdate,
			*lnwire.ChannelAnnouncement,
			*lnwire.NodeAnnouncement,
			*lnwire.AnnounceSignatures,
			*lnwire.GossipTimestampRange,
			*lnwire.QueryShortChanIDs,
			*lnwire.QueryChannelRange,
			*lnwire.ReplyChannelRange,
			*lnwire.ReplyShortChanIDsEnd:

			discStream.AddMsg(msg)

		default:
			peerLog.Errorf("unknown message %v received from peer "+
				"%v", uint16(msg.MsgType()), p)
		}

		if isChanUpdate {
			// If this is a channel update, then we need to feed it
			// into the channel's in-order message stream.
			chanStream, ok := chanMsgStreams[targetChan]
			if !ok {
				// If a stream hasn't yet been created, then
				// we'll do so, add it to the map, and finally
				// start it.
				chanStream = newChanMsgStream(p, targetChan)
				chanMsgStreams[targetChan] = chanStream
				chanStream.Start()
				defer chanStream.Stop()
			}

			// With the stream obtained, add the message to the
			// stream so we can continue processing message.
			chanStream.AddMsg(nextMsg)
		}

		idleTimer.Reset(idleTimeout)
	}

	p.Disconnect(errors.New("read handler closed"))

	peerLog.Tracef("readHandler for peer %v done", p)
}
645485
func (c *Websocket) Subprotocol() *String {
	s := &String{chain: c.chain}
	if c.conn != nil {
		s.value = c.conn.Subprotocol()
	}
	return s
}
95335
func MessageHandler(Incoming <-chan tgbot.MessageWithUpdateID, bot *tgbot.TgBot) {
	for {
		input := <-Incoming

		if input.Msg.Text != nil {
			text := handleMessageText(*input.Msg.Text, input.Msg)
			if text != "" {
				nmsg, err := bot.SimpleSendMessage(input.Msg, text)
				if err != nil {
					fmt.Println(err)
					continue
				}
				fmt.Println(nmsg.String())
			}
		}
	}
}
1247126
func (n *Node) RPCHandler() (*rpc.Server, error) {
	n.lock.RLock()
	defer n.lock.RUnlock()

	if n.inprocHandler == nil {
		return nil, ErrNodeStopped
	}
	return n.inprocHandler, nil
}
45367
func NetHandler(network, addr string, fmtr Format) (Handler, error) {
	conn, err := net.Dial(network, addr)
	if err != nil {
		return nil, err
	}

	return closingHandler{conn, StreamHandler(conn, fmtr)}, nil
}
596151
func (s *Client) NewSubStatter(prefix string) SubStatter {
	var c *Client
	if s != nil {
		c = &Client{
			prefix:  joinPathComp(s.prefix, prefix),
			sender:  s.sender,
			sampler: s.sampler,
		}
	}
	return c
}
64070
func (s *Server) peerHandler() {
	defer s.wg.Done()
	defer s.removeAllPeers()

	for {
		select {
		case peer := <-s.newPeers:
			s.wg.Add(1)
			go s.handleClient(peer)

		case <-s.quit:
			return
		}
	}
}