consumer

Documentation

Index

• Variables
• type Consumer
• type DefaultConsumer
  • func NewDefaultConsumer(conn *amqp091.Connection, queueName string, period time.Duration, ...) (*DefaultConsumer, error)
  • func (d *DefaultConsumer) Close() error
  • func (d *DefaultConsumer) CreateTokensMessagesConsumer(ctx context.Context) (TokensMessagesConsumer, error)
  • func (d *DefaultConsumer) Open() error
• type DefaultService
  • func NewDefaultService(consumer Consumer, tokenValidator gojwttokenclaims.TokenValidator, ...) (*DefaultService, error)
  • func (d *DefaultService) Start(ctx context.Context) error
• type DefaultTokensMessagesConsumer
  • func NewDefaultTokensMessagesConsumer(deliveryCh <-chan amqp091.Delivery, bufferSize int, period time.Duration, ...) (*DefaultTokensMessagesConsumer, error)
  • func (d DefaultTokensMessagesConsumer) ConsumeTokensMessages(ctx context.Context) error
  • func (d DefaultTokensMessagesConsumer) GetChannel() <-chan *gojwtrabbitmq.TokensMessage
• type Service
• type TokensMessagesConsumer

Variables

var (
	// DefaultTickerInterval is the default interval for the ticker in the consumer
	DefaultTickerInterval = 1 * time.Second

	// DefaultTokensMessageConsumerChannelBufferSize is the default buffer size for the tokens message consumer channel
	DefaultTokensMessageConsumerChannelBufferSize = 100
)

var (
	ErrMissingJTI = errors.New("missing jti")
)

Types

type Consumer

type Consumer interface {
	Open() error
	Close() error
	CreateTokensMessagesConsumer(ctx context.Context) (
		TokensMessagesConsumer,
		error,
	)
}

Consumer is the interface for the JWT RabbitMQ consumer

type DefaultConsumer

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

DefaultConsumer is the default implementation of the Consumer interface

func NewDefaultConsumer

func NewDefaultConsumer(
	conn *amqp091.Connection,
	queueName string,
	period time.Duration,
	tokensMessagesConsumerChannelBufferSize int,
	logger *slog.Logger,
) (*DefaultConsumer, error)

NewDefaultConsumer creates a new DefaultConsumer

Parameters:

• conn: the RabbitMQ connection
• queueName: the name of the queue
• period: the polling period
• tokensMessagesConsumerChannelBufferSize: the buffer size for the messages channel
• logger: the logger

Returns:

• *DefaultConsumer: the DefaultConsumer instance
• error: an error if the connection is nil

func (*DefaultConsumer) Close

func (d *DefaultConsumer) Close() error

Close closes the RabbitMQ channel and connection

Returns:

• error: an error if the channel could not be closed

func (*DefaultConsumer) CreateTokensMessagesConsumer

func (d *DefaultConsumer) CreateTokensMessagesConsumer(ctx context.Context) (
	TokensMessagesConsumer,
	error,
)

CreateTokensMessagesConsumer creates a tokens messages consumer

Parameters:

• ctx: the context

Returns:

• TokensMessagesConsumer: the tokens messages consumer
• error: an error if the consumer could not be created

func (*DefaultConsumer) Open

func (d *DefaultConsumer) Open() error

Open opens a RabbitMQ channel

Returns:

• error: an error if the channel could not be opened

type DefaultService

type DefaultService struct {
	gojwttokenclaims.TokenValidator
	// contains filtered or unexported fields
}

DefaultService is the default implementation of the Service interface

func NewDefaultService

func NewDefaultService(
	consumer Consumer,
	tokenValidator gojwttokenclaims.TokenValidator,
	logger *slog.Logger,
) (*DefaultService, error)

NewDefaultService creates a new DefaultService

Parameters:

• handler: the SQL connection handler
• consumer: the RabbitMQ consumer
• tokenValidator: the token validator
• logger: the logger (optional, can be nil)

Returns:

• *DefaultService: the DefaultService instance
• error: an error if the data source or driver name is empty

func (*DefaultService) Start

func (d *DefaultService) Start(ctx context.Context) error

Start starts the service to listen for messages and update the SQL database

Parameters:

• ctx: the context for managing cancellation and timeouts

Returns:

• error: an error if the service could not be started

type DefaultTokensMessagesConsumer

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

DefaultTokensMessagesConsumer is the default implementation of the TokensMessagesConsumer interface

func NewDefaultTokensMessagesConsumer

func NewDefaultTokensMessagesConsumer(
	deliveryCh <-chan amqp091.Delivery,
	bufferSize int,
	period time.Duration,
	logger *slog.Logger,
) (*DefaultTokensMessagesConsumer, error)

NewDefaultTokensMessagesConsumer creates a new DefaultTokensMessagesConsumer

Parameters:

• deliveryCh: the RabbitMQ delivery channel
• bufferSize: the buffer size for the messages channel
• period: the polling period
• logger: the logger

Returns:

• *DefaultTokensMessagesConsumer: the DefaultTokensMessagesConsumer instance
• error: an error if the delivery channel is nil

func (DefaultTokensMessagesConsumer) ConsumeTokensMessages

func (d DefaultTokensMessagesConsumer) ConsumeTokensMessages(
	ctx context.Context,
) error

ConsumeTokensMessages consumes a tokens message and sends it to the channel

Parameters:

• ctx: the context

Returns:

func (DefaultTokensMessagesConsumer) GetChannel

func (d DefaultTokensMessagesConsumer) GetChannel() <-chan *gojwtrabbitmq.TokensMessage

GetChannel returns the tokens message consumer channel

Returns:

• <-chan gojwtrabbitmq.TokensMessage: the tokens message consumer channel

type Service

type Service interface {
	gojwttokenclaims.TokenValidator
	Start(ctx context.Context) error
}

Service is the interface for the SQLite service for JWT IDs

type TokensMessagesConsumer

type TokensMessagesConsumer interface {
	GetChannel() <-chan *gojwtrabbitmq.TokensMessage
	ConsumeTokensMessages(ctx context.Context) error
}

TokensMessagesConsumer is the interface for the JWT RabbitMQ tokens messages consumer