monitor

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Dec 31, 2025 License: MIT Imports: 13 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	WatermillUUIDKey     = "_watermill_message_uuid"
	WatermillPayloadKey  = "payload"
	WatermillMetadataKey = "metadata"
)

Watermill message keys

View Source 
const (
	ReasonPoisonedKey     = "reason_poisoned"
	TopicPoisonedKey      = "topic_poisoned"
	HandlerPoisonedKey    = "handler_poisoned"
	SubscriberPoisonedKey = "subscriber_poisoned"
)

Watermill PoisonQueue metadata keys

Variables

View Source 
var ErrInvalidSortOrder = errors.New("not a valid SortOrder")

Functions

func ParseStreamTimestamp 

func ParseStreamTimestamp(id string) (*time.Time, error)

Types

type DLQMessage 

type DLQMessage struct {
	ID            string         `json:"id"`
	Payload       map[string]any `json:"payload"`
	Timestamp     time.Time      `json:"timestamp"`
	OriginalTopic string         `json:"original_topic"`
	Error         string         `json:"error"`
}

type DLQService 

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

func NewDLQService 

func NewDLQService(monitor *RedisStream, dlqName string) *DLQService

func (*DLQService) DeleteMessage 

func (d *DLQService) DeleteMessage(ctx context.Context, id string) error

func (*DLQService) GetMessage 

func (d *DLQService) GetMessage(ctx context.Context, id string) (*DLQMessage, error)

func (*DLQService) GetMessages 

func (d *DLQService) GetMessages(ctx context.Context, opts PaginationOpts) (*MessageList[DLQMessage], error)

func (*DLQService) GetStats 

func (d *DLQService) GetStats(ctx context.Context) (*StreamInfo, error)

func (*DLQService) RequeueAll 

func (d *DLQService) RequeueAll(ctx context.Context) (int64, error)

func (*DLQService) RequeueMessage 

func (d *DLQService) RequeueMessage(ctx context.Context, id string, payload map[string]any) error

type Message 

type Message struct {
	ID        string         `json:"id"`
	Payload   map[string]any `json:"payload"`
	Timestamp time.Time      `json:"timestamp"`
}

type MessageList 

type MessageList[T any] struct {
	Messages   []T    `json:"messages"`
	TotalCount int64  `json:"total_count"`
	HasMore    bool   `json:"has_more"`
	NextCursor string `json:"next_cursor,omitempty"`
}

type Monitor 

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

func New 

func New(redisClient redis.UniversalClient, dlqName string) *Monitor

func (*Monitor) DLQ 

func (m *Monitor) DLQ() *DLQService

func (*Monitor) GetOverview 

func (m *Monitor) GetOverview(ctx context.Context) (*StatsOverview, error)

func (*Monitor) Streams 

func (m *Monitor) Streams() *StreamService

type PaginationOpts 

type PaginationOpts struct {
	Cursor string
	Limit  int64
	Order  SortOrder
}

func (PaginationOpts) WithDefaults 

func (p PaginationOpts) WithDefaults() PaginationOpts

type RedisStream 

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

func NewRedisStream 

func NewRedisStream(client redis.UniversalClient) *RedisStream

func (*RedisStream) AddMessage 

func (r *RedisStream) AddMessage(ctx context.Context, stream string, data map[string]any) (string, error)

func (*RedisStream) DeleteMessage 

func (r *RedisStream) DeleteMessage(ctx context.Context, stream, id string) error

func (*RedisStream) GetMemoryUsage 

func (r *RedisStream) GetMemoryUsage(ctx context.Context, stream string) (int64, error)

func (*RedisStream) GetStreamInfo 

func (r *RedisStream) GetStreamInfo(ctx context.Context, stream string) (*redis.XInfoStream, error)

func (*RedisStream) GetStreamLength 

func (r *RedisStream) GetStreamLength(ctx context.Context, stream string) (int64, error)

func (*RedisStream) ReadMessage 

func (r *RedisStream) ReadMessage(ctx context.Context, stream, id string) (*redis.XMessage, error)

func (*RedisStream) ReadMessages 

func (r *RedisStream) ReadMessages(ctx context.Context, stream string, opts PaginationOpts) ([]redis.XMessage, error)

func (*RedisStream) ScanStreams 

func (r *RedisStream) ScanStreams(ctx context.Context) ([]string, error)

type SortOrder 

type SortOrder string

ENUM(asc, desc) 

const (
	// SortOrderAsc is a SortOrder of type asc.
	SortOrderAsc SortOrder = "asc"
	// SortOrderDesc is a SortOrder of type desc.
	SortOrderDesc SortOrder = "desc"
)

func ParseSortOrder 

func ParseSortOrder(name string) (SortOrder, error)

ParseSortOrder attempts to convert a string to a SortOrder.

func (*SortOrder) AppendText 

func (x *SortOrder) AppendText(b []byte) ([]byte, error)

AppendText appends the textual representation of itself to the end of b (allocating a larger slice if necessary) and returns the updated slice.

Implementations must not retain b, nor mutate any bytes within b[:len(b)].

func (SortOrder) IsValid 

func (x SortOrder) IsValid() bool

IsValid provides a quick way to determine if the typed value is part of the allowed enumerated values

func (SortOrder) MarshalText 

func (x SortOrder) MarshalText() ([]byte, error)

MarshalText implements the text marshaller method.

func (SortOrder) String 

func (x SortOrder) String() string

String implements the Stringer interface.

func (*SortOrder) UnmarshalText 

func (x *SortOrder) UnmarshalText(text []byte) error

UnmarshalText implements the text unmarshaller method.

type StatsOverview 

type StatsOverview struct {
	TotalStreams     int   `json:"total_streams"`
	TotalMessages    int64 `json:"total_messages"`
	TotalDLQMessages int64 `json:"total_dlq_messages"`
}

type StreamDetail 

type StreamDetail struct {
	StreamInfo
	FirstEntryID *string `json:"first_entry_id,omitempty"`
}

type StreamInfo 

type StreamInfo struct {
	Name         string     `json:"name"`
	Length       int64      `json:"length"`
	MemoryBytes  int64      `json:"memory_bytes"`
	LastEntryID  *string    `json:"last_entry_id,omitempty"`
	LastActivity *time.Time `json:"last_activity"`
}

type StreamService 

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

func NewStreamService 

func NewStreamService(monitor *RedisStream, dlqName string) *StreamService

func (*StreamService) DeleteMessage 

func (s *StreamService) DeleteMessage(ctx context.Context, stream, id string) error

func (*StreamService) GetMessage 

func (s *StreamService) GetMessage(ctx context.Context, stream, id string) (*Message, error)

func (*StreamService) GetStreamDetail 

func (s *StreamService) GetStreamDetail(ctx context.Context, stream string) (*StreamDetail, error)

func (*StreamService) GetStreamMessages 

func (s *StreamService) GetStreamMessages(ctx context.Context, stream string, opts PaginationOpts) (*MessageList[Message], error)

func (*StreamService) GetStreams 

func (s *StreamService) GetStreams(ctx context.Context) ([]StreamInfo, error)

type WatermillMessage 

type WatermillMessage struct {
	UUID     string
	Payload  map[string]any
	Metadata map[string]string
}

func ParseWatermillMessage 

func ParseWatermillMessage(values map[string]any) (*WatermillMessage, error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.
⌂ 换网址