sender

README

Sender usage (producer)

Go Modules:

require github.com/azure-open-tools/event-hubs/sender v1.0.3

Import

import github.com/azure-open-tools/event-hubs/sender

The connections string must be the 'SEND' action with the topic specified.

Simple usage:

builder := sender.NewSenderBuilder()
builder.SetConnectionString("Endpoint://....") //required field 
sender, err := builder.GetSender()

sender.SendMessage("myFirstMessage", context.Background())

Add properties/metadata to your event:

builder := sender.NewSenderBuilder()
builder.SetConnectionString("Endpoint://....") //required field 
builder.AddProperties([]string{"propertyKey1:value1", "propertyKey2:value2", "<propertyKey>:<value>"}) //optional field
sender, err := builder.GetSender()

if err == nil {
    err = sender.SendMessage("myFirstMessage", context.Background())
}

Special fields:

• Base64 is an optional field. if the message string you are passing to SendMessage function is a byte array base64 encoded string, set this field to true. this way the library can convert back to byte array. this is necessary since the official event-hubs library encode to a base64 string before send.

builder.SetBase64(true) //true or false. default value is FALSE

• NumberOfMessages is an optional field. you say to the library the amount of events you want to send at once. it is good to use together with SendBatchMessage.

builder.SetNumberOfMessages(int) //optional field. Here we can set how many messages(events) should be sent.

• Event Handlers. You can make the usage of event handlers to application usage.

builder.SetOnAfterSendMessage(func (event *eventhub.Event) {})
builder.SetOnBeforeSendMessage(func (event *eventhub.Event) {})

• Batch Event Handlers, the send batch message process make usage of the amount of cpu available of the current system. It takes advantage of that to create as many go routines it cans to send events in parallel.

builder.SetOnAfterSendBatchMessage(func (batchSize int, workerIndex int) {})
builder.SetOnBeforeSendBatchMessage(func (batchSizeSent int, workerIndex int){})

• Send batch events

builder := sender.NewSenderBuilder()
builder.SetConnectionString("Endpoint://....") //required field 
builder.AddProperties([]string{"propertyKey1:value1", "propertyKey2:value2", "<propertyKey>:<value>"}) //optional field
builder.SetNumberOfMessages(1000000) //1mi events
sender, err := builder.GetSender()

if err == nil {
    err = sender.SendBatchMessage(message, context.Background())
}

Documentation

Overview

Package arraylist implements the array list.

Structure is not thread safe.

Reference: https://en.wikipedia.org/wiki/List_%28abstract_data_type%29

This package helps you to send messages(events) to Azure Event Hubs.

you can send message, send messages in batch which it allows you to send more events at once (it depends of the listSize of your message/event content and properties). you also can enrich your event with metadata adding properties and handle the event before or after send with event handlers.

See more about Azure Event Hubs

https://azure.microsoft.com/services/event-hubs

Index

• type Builder
  • func NewSenderBuilder() *Builder
  • func (builder *Builder) AddPartitionId(partitionId string) ISenderBuilder
  • func (builder *Builder) AddPartitionIds(partitionIds []string) ISenderBuilder
  • func (builder *Builder) AddProperties(filters []string) ISenderBuilder
  • func (builder *Builder) AddProperty(filter string) ISenderBuilder
  • func (builder *Builder) GetSender() (*Sender, error)
  • func (builder *Builder) SetBase64(is64base bool) ISenderBuilder
  • func (builder *Builder) SetConnectionString(connStr string) ISenderBuilder
  • func (builder *Builder) SetNumberOfMessages(amount int64) ISenderBuilder
  • func (builder *Builder) SetOnAfterSendBatchMessage(handler func(batchSizeSent int, workerIndex int)) ISenderBuilder
  • func (builder *Builder) SetOnAfterSendMessage(handler func(event *eventhub.Event)) ISenderBuilder
  • func (builder *Builder) SetOnBeforeSendBatchMessage(handler func(batchSize int, workerIndex int)) ISenderBuilder
  • func (builder *Builder) SetOnBeforeSendMessage(handler func(event *eventhub.Event)) ISenderBuilder
  • func (builder *Builder) SetRandomMessageSuffix(withSuffix bool) ISenderBuilder
• type Container
• type EventList
• type ISender
• type ISenderBuilder
• type List
  • func New(values ...[]*eventhub.Event) *List
  • func (list *List) Add(values ...[]*eventhub.Event)
  • func (list *List) Clear()
  • func (list *List) Empty() bool
  • func (list *List) Get(index int) ([]*eventhub.Event, bool)
  • func (list *List) Insert(index int, values ...[]*eventhub.Event)
  • func (list *List) Remove(index int)
  • func (list *List) Set(index int, value []*eventhub.Event)
  • func (list *List) Size() int
  • func (list *List) String() string
  • func (list *List) Swap(i, j int)
• type Sender
  • func (sender *Sender) AddProperties(properties map[string]string)
  • func (sender *Sender) SendBatchMessage(message string, ctx context.Context) error
  • func (sender *Sender) SendEventsAsBatch(ctx context.Context, events *[]*eventhub.Event) error
  • func (sender *Sender) SendMessage(message string, ctx context.Context) error

Types

type Builder

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

Builder struct implements ISenderBuilder which handles sender creation.

func NewSenderBuilder

func NewSenderBuilder() *Builder

NewSenderBuilder() creates instance of Builder.

func (*Builder) AddPartitionId

func (builder *Builder) AddPartitionId(partitionId string) ISenderBuilder

AddPartitionId(partitionId string) add single partition. Format expected: "0" (a integer among 0 to 32, it will depends of your Event Hubs settings)

func (*Builder) AddPartitionIds

func (builder *Builder) AddPartitionIds(partitionIds []string) ISenderBuilder

AddPartitionIds(partitionIds []string) add a slice of partitionIds ex: []string{"0", "1", ..., "32"}

func (*Builder) AddProperties

func (builder *Builder) AddProperties(filters []string) ISenderBuilder

AddProperty(filters []string) add a slice of properties ex: []string{"propertyKey1:propertyValue1", "propertyKey2:propertyValue2"}

func (*Builder) AddProperty

func (builder *Builder) AddProperty(filter string) ISenderBuilder

AddProperty(filter string) add a single property to the event, the format expected is: "propertyKey:propertyValue"

func (*Builder) GetSender

func (builder *Builder) GetSender() (*Sender, error)

GetSender() returns the mounted sender structure reference.

func (*Builder) SetBase64

func (builder *Builder) SetBase64(is64base bool) ISenderBuilder

SetBase64(isBase64 bool) set true in case of your message is a binary content. it needs to the sender convert back to the binary format([]byte) before send.

func (*Builder) SetConnectionString

func (builder *Builder) SetConnectionString(connStr string) ISenderBuilder

SetConnectionString(connStr string) Required field. Connection string format is something like: "Endpoint=sb://<namespace>.servicebus.windows.net/;SharedAccessKeyName=send;SharedAccessKey=<AccessKey>;EntityPath=<topic>'"" note SharedAccessKeyName, there are two available values possible: send and listen. To this library must be SEND

func (*Builder) SetNumberOfMessages

func (builder *Builder) SetNumberOfMessages(amount int64) ISenderBuilder

SetNumberOfMessages(amount int64) set the number of messages you want to send. This will repeat the message content and send them according the amount you have set.

func (*Builder) SetOnAfterSendBatchMessage

func (builder *Builder) SetOnAfterSendBatchMessage(handler func(batchSizeSent int, workerIndex int)) ISenderBuilder

SetOnAfterSendBatchMessage(handler func(batchSizeSent int, workerIndex int)) If you wish to see which amount of message was sent by batch you can register to this handle, it will delivery the amount of messages sent by a batch and with worker associated to that batch.

func (*Builder) SetOnAfterSendMessage

func (builder *Builder) SetOnAfterSendMessage(handler func(event *eventhub.Event)) ISenderBuilder

SetOnAfterSendMessage(handler func(event *eventhub.Event)) If you wish to see which event was sent to event hubs, you can register to this handler.

func (*Builder) SetOnBeforeSendBatchMessage

func (builder *Builder) SetOnBeforeSendBatchMessage(handler func(batchSize int, workerIndex int)) ISenderBuilder

SetOnBeforeSendBatchMessage(handler func(batchSize int, workerIndex int)) If you wish to see which amount of batches is about to be sent you can register to this handle, it will delivery the amount of batches by worker associated to that batch. It batches can hold thousand of messages at once.

func (*Builder) SetOnBeforeSendMessage

func (builder *Builder) SetOnBeforeSendMessage(handler func(event *eventhub.Event)) ISenderBuilder

SetOnBeforeSendMessage(handler func(event *eventhub.Event)) If you wish to see which event is about to be sent to event hubs, you can register to this handler.

func (*Builder) SetRandomMessageSuffix

func (builder *Builder) SetRandomMessageSuffix(withSuffix bool) ISenderBuilder

SetRandomMessageSuffix(withSuffix bool) to avoid the message repetition content in usage along side with SetNumberOfMessages with suffix the library will generate 12 random characters and append in the end of the message content. It useful to make sure different messages are being sent to the event hubs. the format delivered will be something like: "message123-asuYRadaIY1d, where "-asuYRadaIY1d" is the suffix random content generated.

type Container

type Container interface {
	Empty() bool
	Size() int
	Clear()
	Values() []*eventhub.Event
}

type EventList

type EventList interface {
	Get(index int) (*eventhub.Event, bool)
	Remove(index int)
	Add(values ...*eventhub.Event)
	Contains(values ...*eventhub.Event) bool
	Swap(index1, index2 int)
	Insert(index int, values ...*eventhub.Event)
	Set(index int, value *eventhub.Event)

	Container
}

List interface that all lists implement

type ISender

type ISender interface {
	AddProperties(properties map[string]string)
	SendMessage(message string, ctx context.Context) error
	SendBatchMessage(message string, ctx context.Context) error
	SendEventsAsBatch(events *[]*eventhub.Event) error
}

ISender defines methods to send operations against azure event hubs.

type ISenderBuilder

type ISenderBuilder interface {
	AddPartitionId(partitionId string) ISenderBuilder
	AddPartitionIds(partitionIds []string) ISenderBuilder
	AddProperty(filter string) ISenderBuilder
	AddProperties(filters []string) ISenderBuilder
	SetBase64(is64base bool) ISenderBuilder
	SetNumberOfMessages(amount int64) ISenderBuilder
	SetRandomMessageSuffix(withSuffix bool) ISenderBuilder
	SetConnectionString(connStr string) ISenderBuilder
	SetOnAfterSendMessage(handler func(event *eventhub.Event)) ISenderBuilder
	SetOnBeforeSendMessage(handler func(event *eventhub.Event)) ISenderBuilder
	SetOnAfterSendBatchMessage(handler func(batchSizeSent int, workerIndex int)) ISenderBuilder
	SetOnBeforeSendBatchMessage(handler func(batchSize int, workerIndex int)) ISenderBuilder
	GetSender() (*Sender, error)
}

ISenderBuilder this interface defines all properties and handlers available to this package functionalities.

type List

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

List holds the elements in a slice

func New

func New(values ...[]*eventhub.Event) *List

New instantiates a new list and adds the passed values, if any, to the list

func (*List) Add

func (list *List) Add(values ...[]*eventhub.Event)

Add appends a value at the end of the list

func (*List) Clear

func (list *List) Clear()

Clear removes all elements from the list.

func (*List) Empty

func (list *List) Empty() bool

Empty returns true if list does not contain any elements.

func (*List) Get

func (list *List) Get(index int) ([]*eventhub.Event, bool)

Get returns the element at index. Second return parameter is true if index is within bounds of the array and array is not empty, otherwise false.

func (*List) Insert

func (list *List) Insert(index int, values ...[]*eventhub.Event)

Insert inserts values at specified index position shifting the value at that position (if any) and any subsequent elements to the right. Does not do anything if position is negative or bigger than list's size Note: position equal to list's size is valid, i.e. append.

func (*List) Remove

func (list *List) Remove(index int)

Remove removes the element at the given index from the list.

func (*List) Set

func (list *List) Set(index int, value []*eventhub.Event)

Set the value at specified index Does not do anything if position is negative or bigger than list's size Note: position equal to list's size is valid, i.e. append.

func (*List) Size

func (list *List) Size() int

Size returns number of elements within the list.

func (*List) String

func (list *List) String() string

String returns a string representation of container

func (*List) Swap

func (list *List) Swap(i, j int)

Swap swaps the two values at the specified positions.

type Sender

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

Sender struct implements ISender interface methods.

func (*Sender) AddProperties

func (sender *Sender) AddProperties(properties map[string]string)

AddProperties(properties map[string]interface{}) can be used to add properties using map format.

func (*Sender) SendBatchMessage

func (sender *Sender) SendBatchMessage(message string, ctx context.Context) error

SendBatchMessage(message string, ctx context.Context) send a message to event hubs in batch. this function should be used together with SetNumberOfMessages and maybe SetMessageSuffix in the case you are not generating your own random content.

func (*Sender) SendEventsAsBatch

func (sender *Sender) SendEventsAsBatch(ctx context.Context, events *[]*eventhub.Event) error

func (*Sender) SendMessage

func (sender *Sender) SendMessage(message string, ctx context.Context) error

SendMessage(message string, ctx context.Context) send a message to event hubs.