server

Documentation

Index

• type Config
  • func DefaultConfig() *Config
• type Handler
  • func NewHandler() *Handler
  • func NewHandlerWithDataDir(dataDir string) *Handler
  • func (h *Handler) Close() error
  • func (h *Handler) Handle(req *protocol.Request) *protocol.Response
  • func (h *Handler) Stats() HandlerStats
• type HandlerStats
• type Partition
• type RequestHandler
• type SecurityHandler
  • func NewSecurityHandler(baseHandler RequestHandler, securityManager *security.Manager, enabled bool) *SecurityHandler
  • func (h *SecurityHandler) GetStats() map[string]int64
  • func (h *SecurityHandler) Handle(req *protocol.Request) *protocol.Response
• type Server
  • func New(config *Config, handler RequestHandler) (*Server, error)
  • func (s *Server) Listener() net.Listener
  • func (s *Server) Start() error
  • func (s *Server) Stats() ServerStats
  • func (s *Server) Stop() error
• type ServerStats
• type TenancyHandler
  • func NewTenancyHandler(baseHandler *Handler, tenancyManager *tenancy.Manager, enabled bool) *TenancyHandler
  • func (h *TenancyHandler) GetStats() map[string]int64
  • func (h *TenancyHandler) Handle(req *protocol.Request) *protocol.Response
• type Topic
• type TopicInfo
• type TopicManager
  • func NewTopicManager(dataDir string) *TopicManager
  • func (tm *TopicManager) Close() error
  • func (tm *TopicManager) CreateTopic(name string, numPartitions uint32) error
  • func (tm *TopicManager) DeleteTopic(name string) error
  • func (tm *TopicManager) GetPartition(topic string, partitionID uint32) (*Partition, error)
  • func (tm *TopicManager) ListTopics() []TopicInfo
  • func (tm *TopicManager) TopicExists(name string) bool

Types

type Config

type Config struct {
	// Network address to bind to (e.g., "localhost:9092")
	Address string

	// Maximum number of concurrent connections
	MaxConnections int

	// Read timeout for connections
	ReadTimeout time.Duration

	// Write timeout for connections
	WriteTimeout time.Duration

	// Idle timeout before closing inactive connections
	IdleTimeout time.Duration

	// Maximum message size
	MaxMessageSize uint32

	// Number of worker goroutines for handling requests
	NumWorkers int

	// Enable TCP keep-alive
	KeepAlive bool

	// Keep-alive period
	KeepAlivePeriod time.Duration
}

Config holds server configuration

func DefaultConfig

func DefaultConfig() *Config

DefaultConfig returns default server configuration

type Handler

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

Handler handles incoming requests

func NewHandler

func NewHandler() *Handler

NewHandler creates a new request handler

func NewHandlerWithDataDir

func NewHandlerWithDataDir(dataDir string) *Handler

NewHandlerWithDataDir creates a new request handler with custom data directory

func (*Handler) Close

func (h *Handler) Close() error

Close closes the handler and releases resources

func (*Handler) Handle

func (h *Handler) Handle(req *protocol.Request) *protocol.Response

Handle handles a request and returns a response

func (*Handler) Stats

func (h *Handler) Stats() HandlerStats

Stats returns handler statistics

type HandlerStats

type HandlerStats struct {
	RequestsHandled int64
	ErrorsHandled   int64
	Uptime          time.Duration
}

HandlerStats holds handler statistics

type Partition

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

Partition represents a single partition with its log

type RequestHandler

type RequestHandler interface {
	Handle(req *protocol.Request) *protocol.Response
}

RequestHandler defines the interface for handling requests

type SecurityHandler

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

SecurityHandler wraps a handler with authentication and authorization

func NewSecurityHandler

func NewSecurityHandler(baseHandler RequestHandler, securityManager *security.Manager, enabled bool) *SecurityHandler

NewSecurityHandler creates a new security-aware handler

func (*SecurityHandler) GetStats

func (h *SecurityHandler) GetStats() map[string]int64

GetStats returns security handler statistics

func (*SecurityHandler) Handle

func (h *SecurityHandler) Handle(req *protocol.Request) *protocol.Response

Handle handles a request with security enforcement

type Server

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

Server represents a TCP server

func New

func New(config *Config, handler RequestHandler) (*Server, error)

New creates a new server

func (*Server) Listener

func (s *Server) Listener() net.Listener

Listener returns the server's listener

func (*Server) Start

func (s *Server) Start() error

Start starts the server

func (*Server) Stats

func (s *Server) Stats() ServerStats

Stats returns server statistics

func (*Server) Stop

func (s *Server) Stop() error

Stop stops the server gracefully

type ServerStats

type ServerStats struct {
	ActiveConnections int64
	TotalRequests     int64
	TotalErrors       int64
	TotalBytesSent    int64
	TotalBytesRecv    int64
	Uptime            time.Duration
}

ServerStats holds server statistics

type TenancyHandler

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

TenancyHandler wraps a handler with multi-tenancy and quota enforcement

func NewTenancyHandler

func NewTenancyHandler(baseHandler *Handler, tenancyManager *tenancy.Manager, enabled bool) *TenancyHandler

NewTenancyHandler creates a new tenancy-aware handler

func (*TenancyHandler) GetStats

func (h *TenancyHandler) GetStats() map[string]int64

GetStats returns tenancy handler statistics

func (*TenancyHandler) Handle

func (h *TenancyHandler) Handle(req *protocol.Request) *protocol.Response

Handle handles a request with tenancy enforcement

type Topic

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

Topic represents a topic with multiple partitions

type TopicInfo

type TopicInfo struct {
	Name          string
	NumPartitions uint32
}

TopicInfo holds information about a topic

type TopicManager

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

TopicManager manages topics and their partitions

func NewTopicManager

func NewTopicManager(dataDir string) *TopicManager

NewTopicManager creates a new topic manager

func (*TopicManager) Close

func (tm *TopicManager) Close() error

Close closes all topic logs

func (*TopicManager) CreateTopic

func (tm *TopicManager) CreateTopic(name string, numPartitions uint32) error

CreateTopic creates a new topic with the specified number of partitions

func (*TopicManager) DeleteTopic

func (tm *TopicManager) DeleteTopic(name string) error

DeleteTopic deletes a topic and all its partitions

func (*TopicManager) GetPartition

func (tm *TopicManager) GetPartition(topic string, partitionID uint32) (*Partition, error)

GetPartition gets a partition for a topic

func (*TopicManager) ListTopics

func (tm *TopicManager) ListTopics() []TopicInfo

ListTopics returns a list of all topics

func (*TopicManager) TopicExists

func (tm *TopicManager) TopicExists(name string) bool

TopicExists checks if a topic exists