sse

README

Perkakas SSE Library

This library helps you to send message to SSE server

How To Send Message

package main

import (
    "context"
    "fmt"
    "github.com/kitabisa/perkakas/v2/sse"
    "github.com/kitabisa/perkakas/v2/queue/kafka"
)

func main() {
    data := map[string]interface{} {
        "donation_id": 5234577,
        "user_id": 267182,
        "amount": 50000,
    }

    // By default, SSE client will set the kafka version to 2.5.0. You can change the kafka version using
    // `SetKafkaVersion()` and see the kafka version currently applied with `GetKafkaVersion()`
    kafkaHost := []string{"localhost:9092"}
    client := sse.NewSseClient(kafkaHost, kafka.WithClientID("katresnan"), kafka.WithRetryMax(5))

    err := client.PublishEvent(context.Background(), "topic", "key", data)
    if err != nil {
        fmt.Println(err)
    }
}

Documentation

Index

• type Client
  • func (s *Client) GetKafkaVersion(ctx context.Context) string
  • func (s *Client) PublishEvent(ctx context.Context, topic string, key string, payload interface{}) (err error)
  • func (s *Client) SetKafkaVersion(ctx context.Context, version string)
• type ISseClient
  • func NewSseClient(host []string, opts ...kafka.ProducerConfigOption) ISseClient

Types

type Client

type Client struct {
	// Host of the kafka brokers.
	Host []string

	// KafkaVersion denotes the expecting kafka version used by this client.
	KafkaVersion string
	// contains filtered or unexported fields
}

Client defines object for SSE instance client

func (*Client) GetKafkaVersion

func (s *Client) GetKafkaVersion(ctx context.Context) string

GetKafkaVersion gets the kafka version

func (*Client) PublishEvent

func (s *Client) PublishEvent(ctx context.Context, topic string, key string, payload interface{}) (err error)

PublishEvent has functionality to publish event to kafka brokers. For payload, you can use marshalable types, such as struct or map[string]interface{}. PublishEvent will publish message to kafka broker in asynchronous fashion.

func (*Client) SetKafkaVersion

func (s *Client) SetKafkaVersion(ctx context.Context, version string)

SetKafkaVersion sets the kafka version

type ISseClient

type ISseClient interface {
	// PublishEvent has functionality to publish event to kafka brokers
	PublishEvent(ctx context.Context, topic string, key string, payload interface{}) (err error)

	// SetKafkaVersion sets the kafka version
	SetKafkaVersion(ctx context.Context, version string)

	// GetKafkaVersion gets the kafka version
	GetKafkaVersion(ctx context.Context) string
}

ISseClient defines interface of SSE client

func NewSseClient

func NewSseClient(host []string, opts ...kafka.ProducerConfigOption) ISseClient

NewSseClient initializes new instance of SSE client