connection

README

Kafka Activity

This connection connects to kafka cluster.

Flogo CLI

flogo install github.com/project-flogo/messaging-contrib/kafka/connection

Configuration

Settings:

Name	Type	Description
brokerUrls	string	The brokers of the Kafka cluster to connect to - REQUIRED
user	string	If connecting to a SASL enabled port, the user id to use for authentication
password	string	If connecting to a SASL enabled port, the password to use for authentication
trustStore	string	If connecting to a TLS secured port, the directory containing the certificates representing the trust chain for the connection. This is usually just the CACert used to sign the server's certificate

Documentation

Index

• type Factory
  • func (*Factory) NewManager(settings map[string]interface{}) (connection.Manager, error)
  • func (*Factory) Type() string
• type KafkaConnect
  • func (c *KafkaConnect) Consumer() interface{}
  • func (c *KafkaConnect) Producer() interface{}
  • func (c *KafkaConnect) Stop() error
• type KafkaConnection
• type KafkaSharedConn
  • func (h *KafkaSharedConn) GetConnection() interface{}
  • func (h *KafkaSharedConn) ReleaseConnection(connection interface{})
  • func (h *KafkaSharedConn) Start() error
  • func (h *KafkaSharedConn) Stop() error
  • func (h *KafkaSharedConn) Type() string
• type Settings

Types

type Factory

type Factory struct {
}

func (*Factory) NewManager

func (*Factory) NewManager(settings map[string]interface{}) (connection.Manager, error)

func (*Factory) Type

func (*Factory) Type() string

type KafkaConnect

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

func (*KafkaConnect) Consumer

func (c *KafkaConnect) Consumer() interface{}

func (*KafkaConnect) Producer

func (c *KafkaConnect) Producer() interface{}

func (*KafkaConnect) Stop

func (c *KafkaConnect) Stop() error

type KafkaConnection

type KafkaConnection interface {
	Producer() interface{}
	Consumer() interface{}
	Stop() error
}

type KafkaSharedConn

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

func (*KafkaSharedConn) GetConnection

func (h *KafkaSharedConn) GetConnection() interface{}

func (*KafkaSharedConn) ReleaseConnection

func (h *KafkaSharedConn) ReleaseConnection(connection interface{})

func (*KafkaSharedConn) Start

func (h *KafkaSharedConn) Start() error

func (*KafkaSharedConn) Stop

func (h *KafkaSharedConn) Stop() error

func (*KafkaSharedConn) Type

func (h *KafkaSharedConn) Type() string

type Settings

type Settings struct {
	BrokerUrls string `md:"brokerUrls,required"` // The Kafka cluster to connect to
	User       string `md:"user"`                // If connecting to a SASL enabled port, the user id to use for authentication
	Password   string `md:"password"`            // If connecting to a SASL enabled port, the password to use for authentication
	TrustStore string `md:"trustStore"`          // If connecting to a TLS secured port, the directory containing the certificates representing the trust chain for the connection. This is usually just the CACert used to sign the server's certificate
}