// Process the information messages from Clients and other Routes.
485398
func (c *client) processInfo(arg []byte) error {
	info := Info{}
	if err := json.Unmarshal(arg, &info); err != nil {
		return err
	}
	switch c.kind {
	case ROUTER:
		c.processRouteInfo(&info)
	case GATEWAY:
		c.processGatewayInfo(&info)
	case LEAF:
		c.processLeafnodeInfo(&info)
	}
	return nil
}
214036
func (s *Server) forwardNewRouteInfoToKnownServers(info *Info) {
	s.mu.Lock()
	defer s.mu.Unlock()

	b, _ := json.Marshal(info)
	infoJSON := []byte(fmt.Sprintf(InfoProto, b))

	for _, r := range s.routes {
		r.mu.Lock()
		if r.route.remoteID != info.ID {
			r.sendInfo(infoJSON)
		}
		r.mu.Unlock()
	}
}
1065476
func (c *Client) processPingreq() error {
	// send a pingresp packet
	err := c.send(packet.NewPingresp(), true)
	if err != nil {
		return c.die(TransportError, err)
	}

	return nil
}
428964
func (nc *Conn) processExpectedInfo() error {

	c := &control{}

	// Read the protocol
	err := nc.readOp(c)
	if err != nil {
		return err
	}

	// The nats protocol should send INFO first always.
	if c.op != _INFO_OP_ {
		return ErrNoInfoReceived
	}

	// Parse the protocol
	if err := nc.processInfo(c.args); err != nil {
		return err
	}

	if nc.Opts.Nkey != "" && nc.info.Nonce == "" {
		return ErrNkeysNotSupported
	}

	return nc.checkForSecure()
}
606484
func (h *eventHub) processEvents() {
	for {
		select {
		case connection := <-h.register:
			h.connections.Register(connection)
		case connection := <-h.unregister:
			h.connections.Unregister(connection)
		case msg := <-h.broadcast:
			h.connections.Broadcast(msg)
		}
	}
}
3168
func (p *Processor) Route(msg interface{}, userData interface{}) error {
	// raw
	if msgRaw, ok := msg.(MsgRaw); ok {
		i, ok := p.msgInfo[msgRaw.msgID]
		if !ok {
			return fmt.Errorf("message %v not registered", msgRaw.msgID)
		}
		if i.msgRawHandler != nil {
			i.msgRawHandler([]interface{}{msgRaw.msgID, msgRaw.msgRawData, userData})
		}
		return nil
	}

	// json
	msgType := reflect.TypeOf(msg)
	if msgType == nil || msgType.Kind() != reflect.Ptr {
		return errors.New("json message pointer required")
	}
	msgID := msgType.Elem().Name()
	i, ok := p.msgInfo[msgID]
	if !ok {
		return fmt.Errorf("message %v not registered", msgID)
	}
	if i.msgHandler != nil {
		i.msgHandler([]interface{}{msg, userData})
	}
	if i.msgRouter != nil {
		i.msgRouter.Go(msgType, msg, userData)
	}
	return nil
}
1331564
func (b *Backend) ProcessEvents(ev ...*auditinternal.Event) bool {
	if b.OnRequest != nil {
		b.OnRequest(ev)
	}
	return true
}
1256258
func (c *client) processInboundLeafMsg(msg []byte) {
	// Update statistics
	c.in.msgs++
	// The msg includes the CR_LF, so pull back out for accounting.
	c.in.bytes += int32(len(msg) - LEN_CR_LF)

	if c.trace {
		c.traceMsg(msg)
	}

	// Check pub permissions
	if c.perms != nil && (c.perms.pub.allow != nil || c.perms.pub.deny != nil) && !c.pubAllowed(string(c.pa.subject)) {
		c.pubPermissionViolation(c.pa.subject)
		return
	}

	srv := c.srv
	acc := c.acc

	// Mostly under testing scenarios.
	if srv == nil || acc == nil {
		return
	}

	// Match the subscriptions. We will use our own L1 map if
	// it's still valid, avoiding contention on the shared sublist.
	var r *SublistResult
	var ok bool

	genid := atomic.LoadUint64(&c.acc.sl.genid)
	if genid == c.in.genid && c.in.results != nil {
		r, ok = c.in.results[string(c.pa.subject)]
	} else {
		// Reset our L1 completely.
		c.in.results = make(map[string]*SublistResult)
		c.in.genid = genid
	}

	// Go back to the sublist data structure.
	if !ok {
		r = c.acc.sl.Match(string(c.pa.subject))
		c.in.results[string(c.pa.subject)] = r
		// Prune the results cache. Keeps us from unbounded growth. Random delete.
		if len(c.in.results) > maxResultCacheSize {
			n := 0
			for subject := range c.in.results {
				delete(c.in.results, subject)
				if n++; n > pruneSize {
					break
				}
			}
		}
	}

	// Check to see if we need to map/route to another account.
	if acc.imports.services != nil {
		c.checkForImportServices(acc, msg)
	}

	// Collect queue names if needed.
	var qnames [][]byte

	// Check for no interest, short circuit if so.
	// This is the fanout scale.
	if len(r.psubs)+len(r.qsubs) > 0 {
		flag := pmrNoFlag
		// If we have queue subs in this cluster, then if we run in gateway
		// mode and the remote gateways have queue subs, then we need to
		// collect the queue groups this message was sent to so that we
		// exclude them when sending to gateways.
		if len(r.qsubs) > 0 && c.srv.gateway.enabled &&
			atomic.LoadInt64(&c.srv.gateway.totalQSubs) > 0 {
			flag = pmrCollectQueueNames
		}
		qnames = c.processMsgResults(acc, r, msg, c.pa.subject, c.pa.reply, flag)
	}

	// Now deal with gateways
	if c.srv.gateway.enabled {
		c.sendMsgToGateways(c.acc, msg, c.pa.subject, c.pa.reply, qnames)
	}
}
1154545
func (proto *Protocol) ProcessData(line string) (reply *Reply) {
	proto.Message.Data += line + "\r\n"

	if strings.HasSuffix(proto.Message.Data, "\r\n.\r\n") {
		proto.Message.Data = strings.Replace(proto.Message.Data, "\r\n..", "\r\n.", -1)

		proto.logf("Got EOF, storing message and switching to MAIL state")
		proto.Message.Data = strings.TrimSuffix(proto.Message.Data, "\r\n.\r\n")
		proto.State = MAIL

		defer proto.resetState()

		if proto.MessageReceivedHandler == nil {
			return ReplyStorageFailed("No storage backend")
		}

		id, err := proto.MessageReceivedHandler(proto.Message)
		if err != nil {
			proto.logf("Error storing message: %s", err)
			return ReplyStorageFailed("Unable to store message")
		}
		return ReplyOk("Ok: queued as " + id)
	}

	return
}
484960
func (s *StanServer) processClientPings(m *nats.Msg) {
	if len(m.Data) == 0 {
		return
	}
	ping := &pb.Ping{}
	if err := ping.Unmarshal(m.Data); err != nil {
		return
	}
	var reply []byte
	client := s.clients.lookupByConnID(ping.ConnID)
	if client != nil {
		// If the client has failed heartbeats and since the
		// server just received a PING from the client, reset
		// the server-to-client HB timer so that a PING is
		// sent soon and the client's subscriptions failedHB
		// is cleared.
		client.RLock()
		hasFailedHBs := client.fhb > 0
		client.RUnlock()
		if hasFailedHBs {
			client.Lock()
			client.hbt.Reset(time.Millisecond)
			client.Unlock()
		}
		if s.pingResponseOKBytes == nil {
			s.pingResponseOKBytes, _ = (&pb.PingResponse{}).Marshal()
		}
		reply = s.pingResponseOKBytes
	} else {
		if s.pingResponseInvalidClientBytes == nil {
			pingError := &pb.PingResponse{
				Error: "client has been replaced or is no longer registered",
			}
			s.pingResponseInvalidClientBytes, _ = pingError.Marshal()
		}
		reply = s.pingResponseInvalidClientBytes
	}
	s.ncs.Publish(m.Reply, reply)
}
154176
func (s *StanServer) processMsg(c *channel) {
	ss := c.ss

	// Since we iterate through them all.
	ss.RLock()
	// Walk the plain subscribers and deliver to each one
	for _, sub := range ss.psubs {
		s.sendAvailableMessages(c, sub)
	}

	// Check the queue subscribers
	for _, qs := range ss.qsubs {
		s.sendAvailableMessagesToQueue(c, qs)
	}
	ss.RUnlock()
}
730083
func (p *Process) Info() (*ProcessInfo, error) {
	p.mtx.Lock()
	defer p.mtx.Unlock()
	return &ProcessInfo{Running: p.running}, nil
}
1184145
func (bot TgBot) ProcessAllMsg(msg Message) {
	msg = bot.cleanMessage(msg)
	// Let's try with and without goroutines here
	for _, c := range bot.ChainConditionals {
		if c.canCall(bot, msg) {
			// c.call(bot, msg)
			go c.call(bot, msg)
			return
		}
		if c.UserInChain(msg) {
			return
		}
	}

	// execlater := make([]ConditionCallStructure, 0)
	executed := false
	for _, v := range bot.TestConditionalFuncs {
		// if nm, ok := v.(NoMessageCall); ok {
		// 	execlater = append(execlater, nm)
		// }
		if v.canCall(bot, msg) {
			executed = true
			v.call(bot, msg)
			// go v.call(bot, msg)
		}
	}

	if !executed {
		for _, f := range bot.NoMessageFuncs {
			f.call(bot, msg)
		}
	}
}
740915
func (r *RPCWrapper) ProcessMessage(req *Request, secret string) bool {

	return r.CheckValidity(req, secret)
}
173965
func Info(ctx context.Context, msg string, args ...interface{}) {
	sendToAppender(ctx, InfoLevel, nil, msg, args...)
}
284801
func (bot TgBot) MessagesHandler(Incoming <-chan MessageWithUpdateID) {
	for {
		input := <-Incoming
		go bot.ProcessAllMsg(input.Msg) // go this or not?
	}
}
1306966
func (h *apiExecutor) Info(ctx context.Context, cmd *InfoRequest) *InfoResponse {
	defer func(started time.Time) {
		apiCommandDurationSummary.WithLabelValues(h.protocol, "info").Observe(time.Since(started).Seconds())
	}(time.Now())

	resp := &InfoResponse{}

	info, err := h.node.Info()
	if err != nil {
		h.node.Log(centrifuge.NewLogEntry(centrifuge.LogLevelError, "error calling info", map[string]interface{}{"error": err.Error()}))
		resp.Error = ErrorInternal
		return resp
	}

	nodes := make([]*NodeResult, len(info.Nodes))
	for i, nd := range info.Nodes {
		res := &NodeResult{
			UID:         nd.UID,
			Version:     nd.Version,
			Name:        nd.Name,
			NumClients:  nd.NumClients,
			NumUsers:    nd.NumUsers,
			NumChannels: nd.NumChannels,
			Uptime:      nd.Uptime,
		}
		if nd.Metrics != nil {
			res.Metrics = &Metrics{
				Interval: nd.Metrics.Interval,
				Items:    nd.Metrics.Items,
			}
		}
		nodes[i] = res
	}
	resp.Result = &InfoResult{
		Nodes: nodes,
	}
	return resp
}