kafka

Documentation

Overview

Package kafka provides a complete API to work with Apache Kafka (defined as Kafka).

Specifications leveraged:

• Topic

Index

• Constants
• type Config
  • func (config *Config) Validate() error
• type Handler
  • func New(ctx context.Context, config Config) (*Handler, error)
  • func (h *Handler) Close() error
  • func (h *Handler) Config() integration.Config
  • func (h *Handler) ConfigMap() map[string]any
  • func (h *Handler) Init() error
  • func (h *Handler) IsReady() bool
  • func (h *Handler) ListActivities() []string
  • func (h *Handler) ListWorkflows() []string
  • func (h *Handler) RegisterWithTopic(w worker.Worker, config topic.Config) error
  • func (h *Handler) String() string

Constants

const Integration string = "kafka"

Integration is the string representation of the Kafka integration.

Types

type Config

type Config struct {

	// Integration represents the common config shared across all integrations.
	Integration integration.Config `json:"integration"`
}

Config is the configuration an end-user should set to configure the Kafka integration.

func (*Config) Validate

func (config *Config) Validate() error

Validate validates the config passed by the end-user. It returns an error if anything critical occured.

type Handler

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

Handler handles the Kafka integration and gives access to related workflows and activities.

func New

func New(ctx context.Context, config Config) (*Handler, error)

New returns a new kafka Handler. It applies the configuration passed by the end-user.

func (*Handler) Close

func (h *Handler) Close() error

Close closes the integration and the specifications registered by the end-user. An error is returned in case the Handler has already been closed.

func (*Handler) Config

func (h *Handler) Config() integration.Config

Config returns the common integration's Config, which is the Config shared by all integrations.

func (*Handler) ConfigMap

func (h *Handler) ConfigMap() map[string]any

ConfigMap transforms the configuration to a map, including a "specifications" key with the configuration for each specification configured.

func (*Handler) Init

func (h *Handler) Init() error

Init initializes the integration and specifications the end-user wants to use. An error is returned in case the Handler has already been initialized.

func (*Handler) IsReady

func (h *Handler) IsReady() bool

IsReady returns if the integration is ready to be consumed by the end-user. The integration must be initialized and must not be closed. This also ensures the specifications leveraged are ready to be consumed. If one of them is not, the integration would not be ready.

func (*Handler) ListActivities

func (h *Handler) ListActivities() []string

ListActivities returns a sorted list of activities' name registered by the integration (and the specifications leveraged if applicable).

func (*Handler) ListWorkflows

func (h *Handler) ListWorkflows() []string

ListWorkflows returns a sorted list of workflows' name registered by the integration (and the specifications leveraged if applicable).

func (*Handler) RegisterWithTopic

func (h *Handler) RegisterWithTopic(w worker.Worker, config topic.Config) error

RegisterWithTopic registers the topic specification within the Kafka integration. End-users will have access to workflows and activities exposed by the topic specification.

func (*Handler) String

func (h *Handler) String() string

String returns the string representation of the integration.