// Receives a multi-part message.
491
func (s *Socket) Recv() (parts [][]byte, err error) {
	parts = make([][]byte, 0)
	for more := true; more; {
		var part []byte
		if part, more, err = s.RecvPart(); err != nil {
			return
		}
		parts = append(parts, part)
	}
	return
}
1205813
func (s *Server) receiveMessage(m Message) error {
	switch obj := m.(type) {
	case *CreateShardMessage:
		f := s.holder.Field(obj.Index, obj.Field)
		if f == nil {
			return fmt.Errorf("local field not found: %s/%s", obj.Index, obj.Field)
		}
		if err := f.AddRemoteAvailableShards(roaring.NewBitmap(obj.Shard)); err != nil {
			return errors.Wrap(err, "adding remote available shards")
		}
	case *CreateIndexMessage:
		opt := obj.Meta
		_, err := s.holder.CreateIndex(obj.Index, *opt)
		if err != nil {
			return err
		}
	case *DeleteIndexMessage:
		if err := s.holder.DeleteIndex(obj.Index); err != nil {
			return err
		}
	case *CreateFieldMessage:
		idx := s.holder.Index(obj.Index)
		if idx == nil {
			return fmt.Errorf("local index not found: %s", obj.Index)
		}
		opt := obj.Meta
		_, err := idx.createField(obj.Field, *opt)
		if err != nil {
			return err
		}
	case *DeleteFieldMessage:
		idx := s.holder.Index(obj.Index)
		if err := idx.DeleteField(obj.Field); err != nil {
			return err
		}
	case *DeleteAvailableShardMessage:
		f := s.holder.Field(obj.Index, obj.Field)
		if err := f.RemoveAvailableShard(obj.ShardID); err != nil {
			return err
		}
	case *CreateViewMessage:
		f := s.holder.Field(obj.Index, obj.Field)
		if f == nil {
			return fmt.Errorf("local field not found: %s", obj.Field)
		}
		_, _, err := f.createViewIfNotExistsBase(obj.View)
		if err != nil {
			return err
		}
	case *DeleteViewMessage:
		f := s.holder.Field(obj.Index, obj.Field)
		if f == nil {
			return fmt.Errorf("local field not found: %s", obj.Field)
		}
		err := f.deleteView(obj.View)
		if err != nil {
			return err
		}
	case *ClusterStatus:
		err := s.cluster.mergeClusterStatus(obj)
		if err != nil {
			return err
		}
	case *ResizeInstruction:
		err := s.cluster.followResizeInstruction(obj)
		if err != nil {
			return err
		}
	case *ResizeInstructionComplete:
		err := s.cluster.markResizeInstructionComplete(obj)
		if err != nil {
			return err
		}
	case *SetCoordinatorMessage:
		return s.cluster.setCoordinator(obj.New)
	case *UpdateCoordinatorMessage:
		s.cluster.updateCoordinator(obj.New)
	case *NodeStateMessage:
		err := s.cluster.receiveNodeState(obj.NodeID, obj.State)
		if err != nil {
			return err
		}
	case *RecalculateCaches:
		s.holder.recalculateCaches()
	case *NodeEvent:
		err := s.cluster.ReceiveEvent(obj)
		if err != nil {
			return errors.Wrapf(err, "cluster receiving NodeEvent %v", obj)
		}
	case *NodeStatus:
		s.handleRemoteStatus(obj)
	}

	return nil
}
947886
func (bot TgBot) SimpleSendMessage(msg Message, text string) (res Message, err error) {
	ressm := bot.SendMessage(msg.Chat.ID, text, nil, nil, nil, nil)
	return splitResultInMessageError(ressm)
}
1214255
func (c *Client) ReceiveMsg() (*pstypes.WebSocketMessage, error) {
	return c.ReceiveMsgTimeout(c.ReadTimeout)
}
728871
func (c *ClientStream) Receive(newMessage func() proto.Message, options ...yarpc.StreamOption) (proto.Message, error) {
	return readFromStream(context.Background(), c.stream, newMessage)
}
586818
func (soc *Socket) RecvMessage(flags Flag) (msg []string, err error) {
	msg = make([]string, 0)
	for {
		s, e := soc.Recv(flags)
		if e == nil {
			msg = append(msg, s)
		} else {
			return msg[0:0], e
		}
		more, e := soc.GetRcvmore()
		if e == nil {
			if !more {
				break
			}
		} else {
			return msg[0:0], e
		}
	}
	return
}
47518
func (c *Conn) Receive() (*Message, error) {
	rv, err := Receive(c.conn, c.buf)
	if err != nil {
		return nil, err
	}
	return &rv, nil
}
706981
func (soc *Socket) RecvMessageBytes(flags Flag) (msg [][]byte, err error) {
	msg = make([][]byte, 0)
	for {
		b, e := soc.RecvBytes(flags)
		if e == nil {
			msg = append(msg, b)
		} else {
			return msg[0:0], e
		}
		more, e := soc.GetRcvmore()
		if e == nil {
			if !more {
				break
			}
		} else {
			return msg[0:0], e
		}
	}
	return
}
900780
func Message(level string, msg string) {
	body := buildBody(level, msg)
	data := body["data"].(map[string]interface{})
	data["body"] = messageBody(msg)

	push(body)
}
617899
func (a *MandrillAPI) MessageSendRaw(rawMessage string, to []string, from Recipient, async bool) ([]SendResponse, error) {
	var response []SendResponse
	if rawMessage == "" {
		return response, errors.New("rawMessage cannot be blank")
	}
	if len(to) <= 0 {
		return response, errors.New("You need at least one recipient in the To array")
	}

	var params map[string]interface{} = make(map[string]interface{})
	params["raw_message"] = rawMessage
	params["from_email"] = from.Email
	params["from_name"] = from.Name
	params["to"] = to
	params["async"] = async
	err := parseMandrillJson(a, messages_send_raw_endpoint, params, &response)
	return response, err
}
966246
func SendMessage(tel, msg string) (uint64, error) {
	omsg := &outgoingMessage{
		tel: tel,
		msg: msg,
	}
	return sendMessage(omsg)
}
228037
func (t *Topic) messagePump() {
	var msg *Message
	var buf []byte
	var err error
	var chans []*Channel
	var memoryMsgChan chan *Message
	var backendChan chan []byte

	// do not pass messages before Start(), but avoid blocking Pause() or GetChannel()
	for {
		select {
		case <-t.channelUpdateChan:
			continue
		case <-t.pauseChan:
			continue
		case <-t.exitChan:
			goto exit
		case <-t.startChan:
		}
		break
	}
	t.RLock()
	for _, c := range t.channelMap {
		chans = append(chans, c)
	}
	t.RUnlock()
	if len(chans) > 0 && !t.IsPaused() {
		memoryMsgChan = t.memoryMsgChan
		backendChan = t.backend.ReadChan()
	}

	// main message loop
	for {
		select {
		case msg = <-memoryMsgChan:
		case buf = <-backendChan:
			msg, err = decodeMessage(buf)
			if err != nil {
				t.ctx.nsqd.logf(LOG_ERROR, "failed to decode message - %s", err)
				continue
			}
		case <-t.channelUpdateChan:
			chans = chans[:0]
			t.RLock()
			for _, c := range t.channelMap {
				chans = append(chans, c)
			}
			t.RUnlock()
			if len(chans) == 0 || t.IsPaused() {
				memoryMsgChan = nil
				backendChan = nil
			} else {
				memoryMsgChan = t.memoryMsgChan
				backendChan = t.backend.ReadChan()
			}
			continue
		case <-t.pauseChan:
			if len(chans) == 0 || t.IsPaused() {
				memoryMsgChan = nil
				backendChan = nil
			} else {
				memoryMsgChan = t.memoryMsgChan
				backendChan = t.backend.ReadChan()
			}
			continue
		case <-t.exitChan:
			goto exit
		}

		for i, channel := range chans {
			chanMsg := msg
			// copy the message because each channel
			// needs a unique instance but...
			// fastpath to avoid copy if its the first channel
			// (the topic already created the first copy)
			if i > 0 {
				chanMsg = NewMessage(msg.ID, msg.Body)
				chanMsg.Timestamp = msg.Timestamp
				chanMsg.deferred = msg.deferred
			}
			if chanMsg.deferred != 0 {
				channel.PutMessageDeferred(chanMsg, chanMsg.deferred)
				continue
			}
			err := channel.PutMessage(chanMsg)
			if err != nil {
				t.ctx.nsqd.logf(LOG_ERROR,
					"TOPIC(%s) ERROR: failed to put msg(%s) to channel(%s) - %s",
					t.name, msg.ID, channel.name, err)
			}
		}
	}

exit:
	t.ctx.nsqd.logf(LOG_INFO, "TOPIC(%s): closing ... messagePump", t.name)
}
199781
func (rm *ReceivedMessage) acknowledge(status uint64) error {
	return rm.receiver.(*receiver).engine().Inject(func() {
		// Deliveries are valid as long as the connection is, unless settled.
		rm.pDelivery.SettleAs(uint64(status))
	})
}
414230
func RequestMessage(level string, r *http.Request, msg string) {
	std.RequestMessage(level, r, msg)
}
146749
func Message(data []byte) *executor.Call {
	return &executor.Call{
		Type: executor.Call_MESSAGE,
		Message: &executor.Call_Message{
			Data: data,
		},
	}
}
1272193
func (r *receiver) Receive(message interface{}) error {
	r.decodeLock.Lock()
	defer r.decodeLock.Unlock()
	if r.decoder == nil {
		r.decoder = msgpack.NewDecoder(r.stream)
		r.decoder.AddExtensions(r.stream.initializeExtensions())
	}

	decodeErr := r.decoder.Decode(message)
	if decodeErr == io.EOF {
		r.stream.Close()
		r.decoder = nil
	}
	return decodeErr
}
337759
func (m *Message) MessagesFilter(filter func(*Message) bool) []*Message {

	messages := make([]*Message, 0, 1)
	if filter(m) {
		messages = append(messages, m)
	}

	if m.HasSubMessage() {
		return append(messages, m.SubMessage.MessagesFilter(filter)...)
	}

	if m.HasParts() {
		for _, part := range m.Parts {
			messages = append(messages, part.MessagesFilter(filter)...)
		}
	}
	return messages
}