transport

package
v0.0.0-...-57d2186 Latest Latest
Warning

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

 Go to latest Published: Jun 3, 2026 License: MIT Imports: 28 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrBufferMemoryExceeded = oops.Errorf("buffered producer: batch memory limit exceeded")
View Source 
var ErrServerOverloaded = status.Error(codes.ResourceExhausted, "server overloaded: max queued requests reached")

Functions

This section is empty.

Types

type AdminServer 

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

AdminServer exposes health checks, pprof, and Prometheus metrics on a dedicated port, separate from business endpoints. This prevents pprof from being exposed to untrusted clients.

func NewAdminServer 

func NewAdminServer(ready *atomic.Bool, reg prometheus.Gatherer) *AdminServer

NewAdminServer creates an admin server with /healthz, /readyz, /debug/pprof/*, and /metrics endpoints.

func (*AdminServer) Addr 

func (s *AdminServer) Addr() string

func (*AdminServer) Start 

func (s *AdminServer) Start(addr string) error

func (*AdminServer) Stop 

func (s *AdminServer) Stop()

type BufferedProducer 

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

BufferedProducer accumulates messages and sends them in batches via a single Produce RPC, amortising the per-RPC cost across many messages.

func NewBufferedProducer 

func NewBufferedProducer(c *Client, topic string, batchSize int, linger time.Duration, opts ...BufferedProducerOption) *BufferedProducer

NewBufferedProducer creates a producer that flushes when either batchSize messages are accumulated, memory limit is reached, or linger time elapses, whichever comes first.

func (*BufferedProducer) Close 

func (p *BufferedProducer) Close(ctx context.Context) error

Close drains any buffered messages, stops the linger timer, and prevents further Send calls. Returns the error from the final flush (or nil).

func (*BufferedProducer) Flush 

func (p *BufferedProducer) Flush(ctx context.Context) error

func (*BufferedProducer) Send 

func (p *BufferedProducer) Send(ctx context.Context, key, value []byte, priority uint32) error

type BufferedProducerOption 

type BufferedProducerOption func(*BufferedProducer)

func WithMaxMemoryBytes 

func WithMaxMemoryBytes(n int) BufferedProducerOption

type Client 

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

func NewClient 

func NewClient(addr string) (*Client, error)

func (*Client) Close 

func (c *Client) Close() error

func (*Client) Commit 

func (c *Client) Commit(ctx context.Context, topic, group string, partition int32, offset int64) error

func (*Client) Consume 

func (c *Client) Consume(ctx context.Context, topic, group string, partition int32, maxBytes int32) ([]*pb.MessageEnvelope, error)

func (*Client) CreateTopic 

func (c *Client) CreateTopic(ctx context.Context, topic string, numPartitions int32, mode pb.ScheduleMode) error

func (*Client) DeleteTopic 

func (c *Client) DeleteTopic(ctx context.Context, topic string) error

func (*Client) GetBalancerExperiencesSeen 

func (c *Client) GetBalancerExperiencesSeen(ctx context.Context) (int64, error)

GetBalancerExperiencesSeen queries the broker for the ExperiencesSeen via the existing GetMetrics RPC. Returns -1 if the broker doesn't have a DQN balancer. This is a best-effort method for warmup early exit.

func (*Client) GetMetrics 

func (c *Client) GetMetrics(ctx context.Context) (*pb.MetricsResponse, error)

func (*Client) ListTopics 

func (c *Client) ListTopics(ctx context.Context) ([]*pb.TopicInfo, error)

func (*Client) Produce 

func (c *Client) Produce(ctx context.Context, topic string, key, value []byte, priority uint32) (int32, int64, error)

func (*Client) ProduceBatch 

func (c *Client) ProduceBatch(ctx context.Context, topic string, msgs []*pb.ProduceMessage) ([]*pb.ProduceResult, error)

ProduceBatch sends a batch of messages in a single RPC. Retries up to produceMaxRetries times on ResourceExhausted with exponential backoff.

func (*Client) SetBalancer 

func (c *Client) SetBalancer(ctx context.Context, algo pb.BalancerAlgorithm, numPartitions int32) error

type HTTPBalancerStatsResponse 

type HTTPBalancerStatsResponse struct {
	ExperiencesSeen int64 `json:"experiences_seen"`
}

type HTTPClient 

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

func NewHTTPClient 

func NewHTTPClient(addr string) *HTTPClient

func (*HTTPClient) Close 

func (c *HTTPClient) Close() error

func (*HTTPClient) Commit 

func (c *HTTPClient) Commit(ctx context.Context, topic, group string, partition int32, offset int64) error

func (*HTTPClient) Consume 

func (c *HTTPClient) Consume(ctx context.Context, topic, group string, partition int32, maxBytes int32) ([]HTTPMessageEnvelope, error)

func (*HTTPClient) CreateTopic 

func (c *HTTPClient) CreateTopic(ctx context.Context, topic string, numPartitions int32, mode string) error

func (*HTTPClient) DeleteTopic 

func (c *HTTPClient) DeleteTopic(ctx context.Context, topic string) error

func (*HTTPClient) GetBalancerStats 

func (c *HTTPClient) GetBalancerStats(ctx context.Context) (int64, error)

func (*HTTPClient) GetMetrics 

func (c *HTTPClient) GetMetrics(ctx context.Context) (*HTTPMetricsResponse, error)

func (*HTTPClient) ListTopics 

func (c *HTTPClient) ListTopics(ctx context.Context) ([]HTTPTopicInfo, error)

func (*HTTPClient) Produce 

func (c *HTTPClient) Produce(ctx context.Context, topic string, key, value []byte, priority uint32) (int, int64, error)

func (*HTTPClient) ProduceBatch 

func (c *HTTPClient) ProduceBatch(ctx context.Context, topic string, records []HTTPProduceRecord) ([]HTTPProduceResult, error)

func (*HTTPClient) SetBalancer 

func (c *HTTPClient) SetBalancer(ctx context.Context, algorithm string, numPartitions int) error

type HTTPCommitRequest 

type HTTPCommitRequest struct {
	Group     string `json:"group"`
	Partition int    `json:"partition"`
	Offset    int64  `json:"offset"`
}

type HTTPConsumeResponse 

type HTTPConsumeResponse struct {
	Messages []HTTPMessageEnvelope `json:"messages"`
}

type HTTPCreateTopicRequest 

type HTTPCreateTopicRequest struct {
	Topic         string `json:"topic"`
	NumPartitions int32  `json:"num_partitions"`
	Mode          string `json:"mode"`
}

type HTTPErrorResponse 

type HTTPErrorResponse struct {
	Error string `json:"error"`
}

type HTTPMessageEnvelope 

type HTTPMessageEnvelope struct {
	Offset      int64  `json:"offset"`
	Key         []byte `json:"key"`
	Value       []byte `json:"value"`
	Priority    uint32 `json:"priority"`
	Timestamp   int64  `json:"timestamp"`
	ProduceTime int64  `json:"produce_time,omitempty"`
	AppendTime  int64  `json:"append_time,omitempty"`
}

type HTTPMetricsResponse 

type HTTPMetricsResponse struct {
	Throughput     float64   `json:"Throughput"`
	PartitionLoads []float64 `json:"PartitionLoads"`
	PredictedLoads []float64 `json:"PredictedLoads"`
	DeliveryRatio  float64   `json:"DeliveryRatio"`
	LoadStdDev     float64   `json:"LoadStdDev"`
	MsgRate        float64   `json:"MsgRate"`
	AvgMsgSize     float64   `json:"AvgMsgSize"`
}

type HTTPProduceRecord 

type HTTPProduceRecord struct {
	Key      []byte `json:"key"`
	Value    []byte `json:"value"`
	Priority uint32 `json:"priority"`
}

type HTTPProduceRequest 

type HTTPProduceRequest struct {
	Records []HTTPProduceRecord `json:"records"`
}

type HTTPProduceResponse 

type HTTPProduceResponse struct {
	Results []HTTPProduceResult `json:"results"`
}

type HTTPProduceResult 

type HTTPProduceResult struct {
	Partition int   `json:"partition"`
	Offset    int64 `json:"offset"`
}

type HTTPServer 

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

func NewHTTPServer 

func NewHTTPServer(b *broker.Broker) *HTTPServer

func (*HTTPServer) Addr 

func (s *HTTPServer) Addr() string

func (*HTTPServer) MaxConnections 

func (s *HTTPServer) MaxConnections() int

func (*HTTPServer) Start 

func (s *HTTPServer) Start(addr string) error

func (*HTTPServer) Stop 

func (s *HTTPServer) Stop()

type HTTPSetBalancerRequest 

type HTTPSetBalancerRequest struct {
	Algorithm     string `json:"algorithm"`
	NumPartitions int    `json:"num_partitions"`
}

type HTTPTopicInfo 

type HTTPTopicInfo struct {
	Topic         string `json:"topic"`
	NumPartitions int32  `json:"num_partitions"`
	Mode          string `json:"mode"`
}

type Server 

type Server struct {
	pb.UnimplementedNyaQueueServer
	// contains filtered or unexported fields
}

func NewServer 

func NewServer(b *broker.Broker) *Server

func (*Server) Addr 

func (s *Server) Addr() string

func (*Server) Commit 

func (s *Server) Commit(_ context.Context, req *pb.CommitRequest) (*pb.CommitResponse, error)

func (*Server) Consume 

func (s *Server) Consume(ctx context.Context, req *pb.ConsumeRequest) (*pb.ConsumeResponse, error)

Consume fetches messages up to maxBytes, tracking the offset locally within the batch. A single Commit at the end avoids per-message offset store writes.

func (*Server) CreateTopic 

func (s *Server) CreateTopic(_ context.Context, req *pb.CreateTopicRequest) (*pb.CreateTopicResponse, error)

func (*Server) DeleteTopic 

func (s *Server) DeleteTopic(_ context.Context, req *pb.DeleteTopicRequest) (*pb.DeleteTopicResponse, error)

func (*Server) GetMetrics 

func (s *Server) GetMetrics(_ context.Context, _ *pb.MetricsRequest) (*pb.MetricsResponse, error)

func (*Server) ListTopics 

func (s *Server) ListTopics(_ context.Context, _ *pb.ListTopicsRequest) (*pb.ListTopicsResponse, error)

func (*Server) Produce 

func (s *Server) Produce(ctx context.Context, req *pb.ProduceRequest) (*pb.ProduceResponse, error)

func (*Server) SetBalancer 

func (s *Server) SetBalancer(_ context.Context, req *pb.SetBalancerRequest) (*pb.SetBalancerResponse, error)

func (*Server) Start 

func (s *Server) Start(addr string) error

func (*Server) Stop 

func (s *Server) Stop()

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.