workqueue

Documentation

Index

• type ChannelJoinHandler
• type ChannelSvc
• type Config
• type Context
• type Event
• type EventMetadata
• type I
  • func New(cfg Config) (*I, error)
  • func (i *I) Publish(e Event, eventTimestamp int64, eventID, requestID string, jsonData []byte) error
  • func (i *I) RegisterChannelJoinsHandler(timeout time.Duration, fn ChannelJoinHandler)
  • func (i *I) RegisterPrivateMessagesHandler(timeout time.Duration, fn MessageHandler)
  • func (i *I) RegisterPublicMessagesHandler(timeout time.Duration, fn MessageHandler)
  • func (i *I) RegisterTeamJoinsHandler(timeout time.Duration, fn TeamJoinHandler)
  • func (i *I) Run()
  • func (i *I) Shutdown()
• type MessageHandler
• type Publisher
• type Q
• type Registerer
• type TeamJoinHandler

Types

type ChannelJoinHandler

type ChannelJoinHandler func(ctx Context, cj *slackevents.MemberJoinedChannelEvent) (shouldRetry, discarded bool, err error)

ChannelJoinHandler is the handler for member_joined_channel Slack events, used when a member joins a channel. For info on nOack please see the comment for the MessageHandler type.

If discarded is true, the returend error isn't treated as an error but instead an informational message.

type ChannelSvc

type ChannelSvc interface {
	Lookup(channelName string) (slack.Channel, bool, error)
}

ChannelSvc is an interface providing the channel service.

type Config

type Config struct {
	// ConsumerName is this node's unique identifier. Leave blank to use
	// hostname.
	ConsumerName string

	// ConsumerGroup is likely this node's application or service name. Leave
	// blank to use hostname, although that's not recommended. If you are only
	// producing events this is safe to be kept blank.
	ConsumerGroup string

	// VisibilityTimeout is how long a consumer will wait for others to finish a
	// task before assuming they are dead and stealing it. If you're acting as
	// only a producer this can be left as its zero value.
	VisibilityTimeout time.Duration

	// RedisClient is the *redis.Client to use for the workqueue.
	RedisClient *redis.Client

	// Logger is the logger
	Logger *zerolog.Logger

	// SlackClient is the client we give to handlers
	SlackClient *slack.Client

	// SlackUser is the slack user that this consumer is running as.
	SlackUser *slack.User

	// ChannelCache is the cache the workqueue will present as the ChannelSvc.
	// Generally this is implemented by a *cache.Channel.
	ChannelCache ChannelSvc
}

Config is the I configuration

type Context

type Context interface {
	context.Context

	// Meta is the event metadata about this job. When did Slack emit it, when
	// did we ingest it, etc.
	Meta() EventMetadata

	// Logger presents the logger create for this handler. This logger isn't
	// global, and is instead request local
	Logger() *zerolog.Logger

	// Slack is the Slack client.
	Slack() *slack.Client

	// Self is the info for the bot user we're using the credentials of.
	Self() slack.User

	// ChannelSvc provides a way to work with the internal channel metadata
	// cache.
	ChannelSvc() ChannelSvc
}

Context is a superset of context.Context, including methods needed by workqueue handler authors. The context given to handlers has a timeout for when they should

type Event

type Event string

Event matches external event types to the Redis stream names we're using

const (
	// SlackMessageChannel is the Event for a message with a channel_type of "channel"
	SlackMessageChannel Event = slackPublicMessage

	// SlackMessageAppHome is the Event for a message with a channel_type of "app_home"
	SlackMessageAppHome Event = slackPrivateMessage

	// SlackMessageGroup is the Event for a message with a channel_type of "group",
	// aka a private channel
	SlackMessageGroup Event = slackPrivateMessage

	// SlackMessageIM is the Event for a message with a channel_type of "im",
	// aka a DM
	SlackMessageIM Event = slackPrivateMessage

	// SlackMessageMPIM is the Event for a message with a channel_type "mpim",
	// aka a group DM
	SlackMessageMPIM Event = slackPrivateMessage

	// SlackTeamJoin is the Event for a team (workspace) join Slack event
	SlackTeamJoin Event = slackTeamJoin

	// SlackChannelJoin is the Event for a channel (public or private) join Slack event.
	SlackChannelJoin Event = slackChannelJoin
)

type EventMetadata

type EventMetadata struct {
	// ID represents the ID as given to us by Slack.
	ID string

	// Time is the time the event was emitted according to Slack.
	Time time.Time

	// IngestTime is the time we received the event from Slack.
	IngestTime time.Time

	// RedisEvent is the ID of the message sent through the Redis queue.
	RedisEvent string
}

EventMetadata represents the metadata about the event

type I

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

I is the workqueue struct, which satisfies Q.

func New

func New(cfg Config) (*I, error)

New returns a new *I or an error. The consumerName, consumerGroup, and visibilityTimeout can be left at their zero value if you're only using I to publish.

func (*I) Publish

func (i *I) Publish(e Event, eventTimestamp int64, eventID, requestID string, jsonData []byte) error

Publish takes an Event, which roughly map to different Slack event types, the event timestamp (from the Slack side),

func (*I) RegisterChannelJoinsHandler

func (i *I) RegisterChannelJoinsHandler(timeout time.Duration, fn ChannelJoinHandler)

RegisterChannelJoinsHandler registers the handler for events related to people joining channels in the Slack workspace.

func (*I) RegisterPrivateMessagesHandler

func (i *I) RegisterPrivateMessagesHandler(timeout time.Duration, fn MessageHandler)

RegisterPrivateMessagesHandler is the method to register a new handler for private Slack messages. This would be those sent to a private channel, a 1-on-1 DM, or a group DM. The timeout argument specifies how long the handler has to complete, before its context is canceled.

func (*I) RegisterPublicMessagesHandler

func (i *I) RegisterPublicMessagesHandler(timeout time.Duration, fn MessageHandler)

RegisterPublicMessagesHandler is the method to register a new handler for public Slack messages. That would be those sent to a public channel. The timeout argument specifies how long the handler has to complete, before its context is canceled.

func (*I) RegisterTeamJoinsHandler

func (i *I) RegisterTeamJoinsHandler(timeout time.Duration, fn TeamJoinHandler)

RegisterTeamJoinsHandler registers the handler for events related to people joining the Slack workspace.

func (*I) Run

func (i *I) Run()

Run wraps the redisqueue.Consumer.Run method

func (*I) Shutdown

func (i *I) Shutdown()

Shutdown wraps the redisqueue.Consumer.Shutdown method

type MessageHandler

type MessageHandler func(ctx Context, me *slackevents.MessageEvent) (shouldRetry, discarded bool, err error)

MessageHandler is the handler for public Slack messages. The handler signals to the workqueue what to do with the item on failure with the shouldRetry bool. If there is an error, and shouldRetry is true, another worker should pick up the work eventually (assuming there are others).

If discarded is true, the returend error isn't treated as an error but instead an informational message.

type Publisher

type Publisher interface {
	Publish(e Event, eventTimestamp int64, eventID, requetID string, jsonData []byte) error
}

Publisher is the interface for the workqueue publish behavior.

type Q

type Q interface {
	Publisher
	Registerer
}

Q is an interface to describe the entirety of the workqueue.

type Registerer

type Registerer interface {
	RegisterTeamJoinsHandler(timeout time.Duration, fn TeamJoinHandler)
	RegisterChannelJoinsHandler(timeout time.Duration, fn ChannelJoinHandler)
	RegisterPublicMessagesHandler(timeout time.Duration, fn MessageHandler)
	RegisterPrivateMessagesHandler(timeout time.Duration, fn MessageHandler)
}

Registerer is the interface for handler registrations within the workqueue.

type TeamJoinHandler

type TeamJoinHandler func(ctx Context, tj *slack.TeamJoinEvent) (shouldRetry, discarded bool, err error)

TeamJoinHandler is the handler for team_join Slack events, used when a new member joins the workspace. For info on shouldRetry please see the comment for the MessageHandler type.

If discarded is true, the returend error isn't treated as an error but instead an informational message.