app

Documentation

Index

• type Diexodos
• type EupalinosHandler
  • func (s *EupalinosHandler) DequeueMessage(ctx context.Context, channelInfo *pb.ChannelInfo) (*pb.Epistello, error)
  • func (s *EupalinosHandler) EnqueueMessage(ctx context.Context, message *pb.Epistello) (*pb.EnqueueResponse, error)
  • func (s *EupalinosHandler) GetQueueLength(ctx context.Context, channelInfo *pb.ChannelInfo) (*pb.QueueLength, error)
  • func (s *EupalinosHandler) LoadStateFromDisk()
  • func (s *EupalinosHandler) SaveStateToDisk()
  • func (s *EupalinosHandler) StartAutoSave()
  • func (s *EupalinosHandler) StartBroadcasting()
  • func (s *EupalinosHandler) StreamQueueUpdates(stream pb.Eupalinos_StreamQueueUpdatesServer) error

Types

type Diexodos

type Diexodos struct {
	LastMessageReceived time.Time
	Name                string
	InternalID          string
	MessageQueue        map[string]pb.InternalEpistello
	MessageUpdateCh     chan pb.MessageUpdate // Channel for task updates to be broadcasted
}

Diexodos represents a task queue

type EupalinosHandler

type EupalinosHandler struct {
	DiexodosMap []*Diexodos // Slice of Diexodos representing different queues
	Config      *config.Config
	pb.UnimplementedEupalinosServer
	// contains filtered or unexported fields
}

EupalinosHandler is the gRPC server handling task queue operations

func (*EupalinosHandler) DequeueMessage

func (s *EupalinosHandler) DequeueMessage(ctx context.Context, channelInfo *pb.ChannelInfo) (*pb.Epistello, error)

DequeueMessage handles message dequeueing from the specified channel

func (*EupalinosHandler) EnqueueMessage

func (s *EupalinosHandler) EnqueueMessage(ctx context.Context, message *pb.Epistello) (*pb.EnqueueResponse, error)

EnqueueMessage handles message enqueueing

func (*EupalinosHandler) GetQueueLength

func (s *EupalinosHandler) GetQueueLength(ctx context.Context, channelInfo *pb.ChannelInfo) (*pb.QueueLength, error)

GetQueueLength returns the length of the queue for the specified channel

func (*EupalinosHandler) LoadStateFromDisk

func (s *EupalinosHandler) LoadStateFromDisk()

LoadStateFromDisk loads the state of the message queues from disk on startup

func (*EupalinosHandler) SaveStateToDisk

func (s *EupalinosHandler) SaveStateToDisk()

SaveStateToDisk saves the state of the message queues to disk

func (*EupalinosHandler) StartAutoSave

func (s *EupalinosHandler) StartAutoSave()

StartAutoSave starts a goroutine to periodically save the state of the message queues to disk

func (*EupalinosHandler) StartBroadcasting

func (s *EupalinosHandler) StartBroadcasting()

StartBroadcasting starts a goroutine to handle broadcasting of task updates to replicas

func (*EupalinosHandler) StreamQueueUpdates

func (s *EupalinosHandler) StreamQueueUpdates(stream pb.Eupalinos_StreamQueueUpdatesServer) error

StreamQueueUpdates handles bidirectional streaming for task updates between Eupalinos pods