// HandleMessage mocks base method
329513
func (m *MockCryptoSetup) HandleMessage(arg0 []byte, arg1 protocol.EncryptionLevel) bool {
	m.ctrl.T.Helper()
	ret := m.ctrl.Call(m, "HandleMessage", arg0, arg1)
	ret0, _ := ret[0].(bool)
	return ret0
}
882417
func HandleServerControlMessage(conn io.Writer, msg Message) error {
	return HandleControlMessage(conn, ws.StateClientSide, msg)
}
691462
func (t *MockSocket) WriteMessage(messageType int, message []byte) error {
	select {
	case expectedMessage := <-t.expectedRequests:
		if asCanonicalString(expectedMessage.sentMessage) == asCanonicalString(message) {
			// Transfer the response into the received messages channel
			for _, response := range expectedMessage.responses {
				t.receivedMessages <- response
			}
		} else {
			fmt.Println("Unexpected response", asCanonicalString(message))
			fmt.Println("Expected ", string(expectedMessage.sentMessage))
		}
	default:
		fmt.Println("No more responses registered, expecting", string(message))
	}

	return nil
}
545335
func (drv *Driver) wrapMessageHandler(handler MQTTMessageHandler) MQTTMessageHandler {
	return func(msg MQTTMessage) {
		// if XXXStopIt {
		// 	return //////
		// }
		drv.HandleMessageSync(func() {
			handler(msg)
		})
	}
}
980278
func (dht *IpfsDHT) handleNewMessage(s inet.Stream) bool {
	ctx := dht.ctx

	cr := ctxio.NewReader(ctx, s) // ok to use. we defer close stream in this func
	cw := ctxio.NewWriter(ctx, s) // ok to use. we defer close stream in this func
	r := ggio.NewDelimitedReader(cr, inet.MessageSizeMax)
	w := newBufferedDelimitedWriter(cw)
	mPeer := s.Conn().RemotePeer()

	for {
		var req pb.Message
		switch err := r.ReadMsg(&req); err {
		case io.EOF:
			return true
		default:
			// This string test is necessary because there isn't a single stream reset error
			// instance	in use.
			if err.Error() != "stream reset" {
				logger.Debugf("error reading message: %#v", err)
			}
			stats.RecordWithTags(
				ctx,
				[]tag.Mutator{tag.Upsert(metrics.KeyMessageType, "UNKNOWN")},
				metrics.ReceivedMessageErrors.M(1),
			)
			return false
		case nil:
		}

		startTime := time.Now()
		ctx, _ = tag.New(
			ctx,
			tag.Upsert(metrics.KeyMessageType, req.GetType().String()),
		)

		stats.Record(
			ctx,
			metrics.ReceivedMessages.M(1),
			metrics.ReceivedBytes.M(int64(req.Size())),
		)

		handler := dht.handlerForMsgType(req.GetType())
		if handler == nil {
			stats.Record(ctx, metrics.ReceivedMessageErrors.M(1))
			logger.Warningf("can't handle received message of type %v", req.GetType())
			return false
		}

		resp, err := handler(ctx, mPeer, &req)
		if err != nil {
			stats.Record(ctx, metrics.ReceivedMessageErrors.M(1))
			logger.Debugf("error handling message: %v", err)
			return false
		}

		dht.updateFromMessage(ctx, mPeer, &req)

		if resp == nil {
			continue
		}

		// send out response msg
		err = w.WriteMsg(resp)
		if err == nil {
			err = w.Flush()
		}
		if err != nil {
			stats.Record(ctx, metrics.ReceivedMessageErrors.M(1))
			logger.Debugf("error writing response: %v", err)
			return false
		}

		elapsedTime := time.Since(startTime)
		latencyMillis := float64(elapsedTime) / float64(time.Millisecond)
		stats.Record(ctx, metrics.InboundRequestLatency.M(latencyMillis))
	}
}
52848
func (e *engineImpl) handlePubSubMessage(c context.Context, msg *pubsub.PubsubMessage) error {
	logging.Infof(c, "Received PubSub message %q", msg.MessageId)

	// Extract Job and Invocation ID from validated auth_token.
	var jobID string
	var invID int64
	data, err := pubsubAuthToken.Validate(c, msg.Attributes["auth_token"], nil)
	if err != nil {
		logging.WithError(err).Errorf(c, "Bad auth_token attribute")
		return err
	}
	jobID = data["job"]
	if invID, err = strconv.ParseInt(data["inv"], 10, 64); err != nil {
		logging.WithError(err).Errorf(c, "Could not parse 'inv' %q", data["inv"])
		return err
	}

	// Hand the message to the controller.
	action := fmt.Sprintf("pubsub message %q", msg.MessageId)
	return e.withController(c, jobID, invID, action, func(c context.Context, ctl *taskController) error {
		err := ctl.manager.HandleNotification(c, ctl, msg)
		switch {
		case err == nil:
			return nil // success! save the invocation
		case transient.Tag.In(err) || tq.Retry.In(err):
			return err // ask for redelivery on transient errors, don't touch the invocation
		}
		// On fatal errors, move the invocation to failed state (if not already).
		if ctl.State().Status != task.StatusFailed {
			ctl.DebugLog("Fatal error when handling PubSub notification, aborting invocation - %s", err)
			ctl.State().Status = task.StatusFailed
		}
		return nil // need to save the invocation, even on fatal errors
	})
}
743580
func (e *Executor) handleMessageStream(ctx context.Context, safeSession *SafeSession, sql string, target querypb.Target, callback func(*sqltypes.Result) error, vcursor *vcursorImpl, logStats *LogStats) error {
	stmt, err := sqlparser.Parse(sql)
	if err != nil {
		logStats.Error = err
		return err
	}

	streamStmt, ok := stmt.(*sqlparser.Stream)
	if !ok {
		logStats.Error = err
		return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unrecognized STREAM statement: %v", sql)
	}

	// TODO: Add support for destination target in streamed queries
	table, _, _, _, err := vcursor.FindTable(streamStmt.Table)
	if err != nil {
		logStats.Error = err
		return err
	}

	execStart := time.Now()
	logStats.PlanTime = execStart.Sub(logStats.StartTime)

	err = e.MessageStream(ctx, table.Keyspace.Name, target.Shard, nil, table.Name.CompliantName(), callback)
	logStats.Error = err
	logStats.ExecuteTime = time.Since(execStart)
	return err
}
414230
func RequestMessage(level string, r *http.Request, msg string) {
	std.RequestMessage(level, r, msg)
}
519513
func handleDataMessage(src string, timestamp uint64, dm *textsecure.DataMessage) error {
	flags, err := handleFlags(src, dm)
	if err != nil {
		return err
	}

	atts, err := handleAttachments(dm)
	if err != nil {
		return err
	}

	gr, err := handleGroups(src, dm)
	if err != nil {
		return err
	}

	msg := &Message{
		source:      src,
		message:     dm.GetBody(),
		attachments: atts,
		group:       gr,
		timestamp:   timestamp,
		flags:       flags,
	}

	if client.MessageHandler != nil {
		client.MessageHandler(msg)
	}
	return nil
}
1093154
func (m TaskManager) HandleNotification(c context.Context, ctl task.Controller, msg *pubsub.PubsubMessage) error {
	return errors.New("not implemented")
}
352719
func (m *MockResolver) HandlePUEvent(ctx context.Context, puID string, event common.Event, runtime policy.RuntimeReader) error {
	ret := m.ctrl.Call(m, "HandlePUEvent", ctx, puID, event, runtime)
	ret0, _ := ret[0].(error)
	return ret0
}
688128
func handleMessages(r io.Reader, stdout io.Writer, stderr io.Writer) error {
	headerBuf := make([]byte, 8)
	for {
		n, e := r.Read(headerBuf)
		if e != nil {
			if e == io.EOF {
				return nil
			}
			return e
		}
		if n != 8 {
			return fmt.Errorf("failed reading; header to short")
		}

		msgLength := messageLength(headerBuf)
		msgBuf := make([]byte, msgLength) // buffer size taken from io.Copy
		n = 0
		for n < msgLength {
			i, e := r.Read(msgBuf[n:])
			if e != nil {
				return e
			}
			n += i
		}

		switch headerBuf[0] {
		case 0: // stdin
			if stdout != nil {
				_, _ = stdout.Write([]byte{'+'})
			}
		case 1: // stdout
			if stdout != nil {
				_, e := stdout.Write(msgBuf[:msgLength])
				if e != nil {
					return e
				}
			}
		case 2: // stderr
			if stderr != nil {
				_, e := stderr.Write(msgBuf[:msgLength])
				if e != nil {
					return e
				}
			}
		default:
			return fmt.Errorf("unknown stream source received")
		}
	}
}
1030585
func (eh messageReactionRemoveEventHandler) Handle(s *Session, i interface{}) {
	if t, ok := i.(*MessageReactionRemove); ok {
		eh(s, t)
	}
}
789825
func (l *Logger) HandleEvent(ctx context.Context, event eh.Event) error {
	log.Println("event:", event)
	return nil
}
78072
func (conn *Conn) handleCall(msg *Message) {
	name := msg.Headers[FieldMember].value.(string)
	path := msg.Headers[FieldPath].value.(ObjectPath)
	ifaceName, _ := msg.Headers[FieldInterface].value.(string)
	sender, hasSender := msg.Headers[FieldSender].value.(string)
	serial := msg.serial
	if ifaceName == "org.freedesktop.DBus.Peer" {
		switch name {
		case "Ping":
			conn.sendReply(sender, serial)
		case "GetMachineId":
			conn.sendReply(sender, serial, conn.uuid)
		default:
			conn.sendError(ErrMsgUnknownMethod, sender, serial)
		}
		return
	}
	if len(name) == 0 {
		conn.sendError(ErrMsgUnknownMethod, sender, serial)
	}

	object, ok := conn.handler.LookupObject(path)
	if !ok {
		conn.sendError(ErrMsgNoObject, sender, serial)
		return
	}

	iface, exists := object.LookupInterface(ifaceName)
	if !exists {
		conn.sendError(ErrMsgUnknownInterface, sender, serial)
		return
	}

	m, exists := iface.LookupMethod(name)
	if !exists {
		conn.sendError(ErrMsgUnknownMethod, sender, serial)
		return
	}
	args, err := conn.decodeArguments(m, sender, msg)
	if err != nil {
		conn.sendError(err, sender, serial)
		return
	}

	ret, err := m.Call(args...)
	if err != nil {
		conn.sendError(err, sender, serial)
		return
	}

	if msg.Flags&FlagNoReplyExpected == 0 {
		reply := new(Message)
		reply.Type = TypeMethodReply
		reply.serial = conn.getSerial()
		reply.Headers = make(map[HeaderField]Variant)
		if hasSender {
			reply.Headers[FieldDestination] = msg.Headers[FieldSender]
		}
		reply.Headers[FieldReplySerial] = MakeVariant(msg.serial)
		reply.Body = make([]interface{}, len(ret))
		for i := 0; i < len(ret); i++ {
			reply.Body[i] = ret[i]
		}
		reply.Headers[FieldSignature] = MakeVariant(SignatureOf(reply.Body...))

		conn.sendMessage(reply)
	}
}
873762
func (c *Conversation) injectMessage(vm ValidMessage) {
	c.injections.messages = append(c.injections.messages, vm)
}
1306079
func ProcessMessage(reader io.Reader, f func([]byte) error) (err error) {
	return TakeMessage(reader, -1, f)
}