Function pluginFireHose

server/plugins.go:353–424

(sess *Session, msg *ClientComMessage)

}

func pluginFireHose(sess *Session, msg *ClientComMessage) (*ClientComMessage, *ServerComMessage) {
	if globals.plugins == nil {
		// Return the original message to continue processing without changes
		return msg, nil
	}

	var req *pbx.ClientReq

	id, topic := pluginIDAndTopic(msg)
	ts := time.Now().UTC().Round(time.Millisecond)
	for i := range globals.plugins {
		p := &globals.plugins[i]
		if !pluginDoFiltering(p.filterFireHose, msg) {
			// Plugin is not interested in FireHose
			continue
		}

		if req == nil {
			// Generate request only if needed
			req = pluginGenerateClientReq(sess, msg)
			if req == nil {
				// Failed to serialize message. Most likely the message is invalid.
				break
			}
		}

		var ctx context.Context
		var cancel context.CancelFunc
		if p.timeout > 0 {
			ctx, cancel = context.WithTimeout(context.Background(), p.timeout)
			defer cancel()
		} else {
			ctx = context.Background()
		}
		if resp, err := p.client.FireHose(ctx, req); err == nil {
			respStatus := resp.GetStatus()
			// CONTINUE means default processing
			if respStatus == pbx.RespCode_CONTINUE {
				continue
			}
			// DROP means stop processing of the message
			if respStatus == pbx.RespCode_DROP {
				return nil, nil
			}
			// REPLACE: ClientMsg was updated by the plugin. Use the new one for further processing.
			if respStatus == pbx.RespCode_REPLACE {
				return pbCliDeserialize(resp.GetClmsg()), nil
			}

			// RESPOND: Plugin provided an alternative response message. Use it
			return nil, pbServDeserialize(resp.GetSrvmsg())

		} else if p.failureCode != 0 {
			// Plugin failed and it's configured to stop further processing.
			logs.Err.Println("plugin: failed,", p.name, err)
			return nil, &ServerComMessage{
				Ctrl: &MsgServerCtrl{
					Id:        id,