amqp

README

---

title: AMQP weight: 15709

amqp

This trigger provides your flogo application the ability to start a flow via AMQP

Installation

flogo install https://github.com/mtorre-iot/flogo-contrib/trigger/amqp

Schema

Settings, Outputs and Endpoint:

{
  "settings":[
    {
      "name": "requestHostName",
      "type": "string",
      "required": true
    },
    {
      "name": "requestPort",
      "type": "string",
      "required": true
    },
    {
      "name": "requestExchangeName",
      "type": "string",
      "required": true
    },
    {
      "name": "requestExchangeType",
      "type": "string",
      "required": true
    },
    {
      "name": "requestRoutingKey",
      "type": "string",
      "required": true
    },
    {
      "name": "requestUser",
      "type": "string",
      "required": true
    },    
    {
      "name": "requestPassword",
      "type": "string",
      "required": true
    },
    {
      "name": "requestDurable",
      "type": "string",
      "required": false
    },
    {
      "name": "requestAutoDelete",
      "type": "string",
      "required": false
    },
    {
      "name": "requestReliable",
      "type": "string",
      "required": false
    },
    {
      "name": "responseHostName",
      "type": "string",
      "required": false
    },
    {
      "name": "responsePort",
      "type": "string",
      "required": false
    },
    {
      "name": "responseExchangeName",
      "type": "string",
      "required": false
    },
    {
      "name": "responseExchangeType",
      "type": "string",
      "required": false
    },
    {
      "name": "responseRoutingKey",
      "type": "string",
      "required": false
    },
    {
      "name": "responseUser",
      "type": "string",
      "required": false
    },    
    {
      "name": "responsePassword",
      "type": "string",
      "required": false
    },
    {
      "name": "responseDurable",
      "type": "string",
      "required": false
    },
    {
      "name": "responseAutoDelete",
      "type": "string",
      "required": false
    },
    {
      "name": "responseReliable",
      "type": "string",
      "required": false
    }
  ],
  "output": [
    {
      "name": "message",
      "type": "string"
    }
  ],
  "reply": [
    {
      "name": "data",
      "type": "object"
    }
  ],
  "handler": {
    "settings": [
      {
        "name": "topic",
        "type": "string",
        "required": true
      }
    ]
  }
}

Example Configurations

Triggers are configured via the triggers.json of your application. The following are some example configuration of the AMQP Trigger.

Start a flow

Configure the Trigger to start "myflow". "settings" "topic" is the topic it uses to listen for incoming messages. So in this case the "endpoints" "settings" "topic" is "test_start" will start "myflow" flow. The incoming message payload has to define "replyTo" which is the the topic used to reply on.

{
  "triggers": [
    {
      "id": "receive_amqp_message",
      "ref": "github.com/mtorre-iot/flogo-contrib/trigger/amqp",
      "name": "Receive AMQP Message",
      "description": "Simple AMQP Trigger",
      "settings": {
        "requestHostName": "localhost",
        "requestPort": "5672",
        "requestExchangeName": "AMQPRequestExchange",
        "requestExchangeType": "topic",
        "requestRoutingKey": "#",
        "requestUser": "guest",
        "requestPassword": "guest",
        "responseHostName": "localhost",
        "responsePort": "5672",
        "responseExchangeName": "AMQPResponseExchange",
        "responseExchangeType": "topic",
        "responseRoutingKey": "#",
        "responseUser": "guest",
        "responsePassword": "guest"
      },
      "handlers": [
        {
          "action": {
            "ref": "github.com/TIBCOSoftware/flogo-contrib/action/flow",
            "data": {
              "flowURI": "res://flow:amqp_application_test"
            },
            "mappings": {
              "input": [
                {
                  "mapTo": "message",
                  "type": "assign",
                  "value": "$.message"
                }
              ]
            }
          },
          "settings": {
            "topic": "update"
          }
        }
      ]
    }
  ]
}

Documentation

Index

• func NewFactory(md *trigger.Metadata) trigger.Factory
• type AMQPExchange
  • func AMQPExchangeNew(hostName string, port int, exchangeName string, exchangeType string, ...) *AMQPExchange
  • func (exch *AMQPExchange) Close() error
  • func (exch *AMQPExchange) Open(isQueued bool) error
  • func (exch *AMQPExchange) PrepareReceiveFunc(f func(msgs <-chan amqp.Delivery)) error
  • func (exch *AMQPExchange) Publish(body string) error
  • func (exch *AMQPExchange) PublishObject(obj interface{}) error
  • func (exch *AMQPExchange) ReadMessages() ([]string, error)
• type AMQPFactory
  • func (t *AMQPFactory) New(config *trigger.Config) trigger.Trigger
• type AmqpTrigger
  • func (t *AmqpTrigger) Initialize(ctx trigger.InitContext) error
  • func (t *AmqpTrigger) Metadata() *trigger.Metadata
  • func (t *AmqpTrigger) RunHandler(handler *trigger.Handler, payload string)
  • func (t *AmqpTrigger) Start() error
  • func (t *AmqpTrigger) Stop() error

Functions

func NewFactory

func NewFactory(md *trigger.Metadata) trigger.Factory

NewFactory create a new Trigger factory

Types

type AMQPExchange

type AMQPExchange struct {
	Uri          string
	HostName     string
	ExchangeName string
	ExchangeType string
	QueueName    string
	RoutingKey   string
	UserName     string
	Password     string
	Durable      bool
	AutoDelete   bool
	Reliable     bool
	Connection   *amqp.Connection
	Channel      *amqp.Channel
	Queue        *amqp.Queue
	Messages     []string
	Confirms     chan amqp.Confirmation
	IsOpen       bool
}

AMQPExchange contains all parameters required to create or open an exchenge

func AMQPExchangeNew

func AMQPExchangeNew(hostName string, port int, exchangeName string, exchangeType string, queueName string, routingKey string,
	userName string, password string, durable bool, autoDelete bool, reliable bool) *AMQPExchange

AMQPExchangeNew creates a new exchange in the Broker

func (*AMQPExchange) Close

func (exch *AMQPExchange) Close() error

Close closes the exchange

func (*AMQPExchange) Open

func (exch *AMQPExchange) Open(isQueued bool) error

Open opens a previosly created Exchange

func (*AMQPExchange) PrepareReceiveFunc

func (exch *AMQPExchange) PrepareReceiveFunc(f func(msgs <-chan amqp.Delivery)) error

PrepareReceiveFunc Prepares exchange/queue to receive messages

func (*AMQPExchange) Publish

func (exch *AMQPExchange) Publish(body string) error

Publish publishes a new message into an existing exchange

func (*AMQPExchange) PublishObject

func (exch *AMQPExchange) PublishObject(obj interface{}) error

PublishObject serializes objects (JSON) and publish to existing exchange

func (*AMQPExchange) ReadMessages

func (exch *AMQPExchange) ReadMessages() ([]string, error)

ReadMessages reads the messajes accumulated in the queue

type AMQPFactory

type AMQPFactory struct {
	// contains filtered or unexported fields
}

AMQPFactory AMQP Trigger factory

func (*AMQPFactory) New

func (t *AMQPFactory) New(config *trigger.Config) trigger.Trigger

New Creates a new trigger instance for a given id

type AmqpTrigger

type AmqpTrigger struct {
	// contains filtered or unexported fields
}

AmqpTrigger is simple AMQP trigger

func (*AmqpTrigger) Initialize

func (t *AmqpTrigger) Initialize(ctx trigger.InitContext) error

Initialize implements trigger.Initializable.Initialize

func (*AmqpTrigger) Metadata

func (t *AmqpTrigger) Metadata() *trigger.Metadata

Metadata implements trigger.Trigger.Metadata

func (*AmqpTrigger) RunHandler

func (t *AmqpTrigger) RunHandler(handler *trigger.Handler, payload string)

RunHandler runs the handler and associated action

func (*AmqpTrigger) Start

func (t *AmqpTrigger) Start() error

Start implements trigger.Trigger.Start

func (*AmqpTrigger) Stop

func (t *AmqpTrigger) Stop() error

Stop implements ext.Trigger.Stop