morpc

type Backend interface {
	// Send send the request for future to the corresponding backend.
	// moerr.ErrBackendClosed returned if backend is closed.
	Send(ctx context.Context, request Message) (*Future, error)
	// SendInternal is similar to Send, but perform on internal message
	SendInternal(ctx context.Context, request Message) (*Future, error)
	// NewStream create a stream used to asynchronous stream of sending and receiving messages.
	// If the underlying connection is reset during the duration of the stream, then the stream
	// will be closed.
	NewStream(unlockAfterClose bool) (Stream, error)
	// Close close the backend.
	Close()
	// Busy the backend receives a lot of requests concurrently during operation, but when the number
	// of requests waiting to be sent reaches some threshold, the current backend is busy.
	Busy() bool
	// LastActiveTime returns last active time
	LastActiveTime() time.Time
	// Lock other I/O operations can not use this backend if the backend is locked
	Lock()
	// Unlock the backend can used by other I/O operations after unlock
	Unlock()
	// Locked indicates if backend is locked
	Locked() bool
}

type BackendFactory interface {
	// Create create the corresponding backend based on the given address.
	Create(address string, extraOptions ...BackendOption) (Backend, error)
}

type BackendOption func(*remoteBackend)

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

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

type CircuitBreakerConfig struct {
	// Enabled determines if circuit breaker is active.
	Enabled bool
	// FailureThreshold is the number of consecutive failures before opening the circuit.
	FailureThreshold int32
	// ResetTimeout is the duration to wait before transitioning from open to half-open.
	ResetTimeout time.Duration
	// HalfOpenMaxRequests is the maximum number of requests allowed in half-open state.
	HalfOpenMaxRequests int32
}

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

type CircuitBreakerStats struct {
	State        CircuitState
	Failures     int32
	LastFailure  time.Time
	OpenCount    int64
	RejectCount  int64
	SuccessCount int64
	FailureCount int64
}

type CircuitState int32

type ClientOption func(*client)

type ClientSession interface {
	// Close close the client session
	Close() error
	// SessionCtx get the session context, if session is closed the context will be canceled.
	SessionCtx() context.Context
	// Write writing the response message to the client.
	Write(ctx context.Context, response Message) error
	// AsyncWrite only put message into write queue, and return immediately.
	AsyncWrite(response Message) error
	// CreateCache create a message cache using cache ID. Cache will removed if
	// context is done.
	CreateCache(ctx context.Context, cacheID uint64) (MessageCache, error)
	// DeleteCache delete cache using the spec cacheID
	DeleteCache(cacheID uint64)
	// GetCache returns the message cache
	GetCache(cacheID uint64) (MessageCache, error)
	// RemoteAddress returns remote address, include ip and port
	RemoteAddress() string
}

type Codec interface {
	codec.Codec
	// Valid valid the message is valid
	Valid(message Message) error
	// AddHeaderCodec add header codec. The HeaderCodecs are added sequentially and the headercodecs are
	// executed in the order in which they are added at codec time.
	AddHeaderCodec(HeaderCodec)
}

type CodecOption func(*messageCodec)

type Config struct {
	// MaxConnections maximum number of connections to communicate with each DNStore.
	// Default is 400.
	MaxConnections int `toml:"max-connections"`
	// MaxIdleDuration maximum connection idle time, connection will be closed automatically
	// if this value is exceeded. Default is 1 min.
	MaxIdleDuration toml.Duration `toml:"max-idle-duration"`
	// SendQueueSize maximum capacity of the send request queue per connection, when the
	// queue is full, the send request will be blocked. Default is 10240.
	SendQueueSize int `toml:"send-queue-size"`
	// BusyQueueSize when the length of the send queue reaches the currently set value, the
	// current connection is busy with high load. When any connection with Busy status exists,
	// a new connection will be created until the value set by MaxConnections is reached.
	// Default is 3/4 of SendQueueSize.
	BusyQueueSize int `toml:"busy-queue-size"`
	// WriteBufferSize buffer size for write messages per connection. Default is 1kb
	WriteBufferSize toml.ByteSize `toml:"write-buffer-size"`
	// ReadBufferSize buffer size for read messages per connection. Default is 1kb
	ReadBufferSize toml.ByteSize `toml:"read-buffer-size"`
	// MaxMessageSize max message size for rpc. Default is 100M
	MaxMessageSize toml.ByteSize `toml:"max-message-size"`
	// PayloadCopyBufferSize buffer size for copy payload to socket. Default is 16kb
	PayloadCopyBufferSize toml.ByteSize `toml:"payload-copy-buffer-size"`
	// EnableCompress enable compress message
	EnableCompress bool `toml:"enable-compress"`

	// ServerWorkers number of server workers for handle requests
	ServerWorkers int `toml:"server-workers"`
	// ServerBufferQueueSize queue size for server buffer requetsts
	ServerBufferQueueSize int `toml:"server-buffer-queue-size"`

	// GCIdleCheckInterval interval for global GC manager to check idle backends.
	// Default is 1 second. This controls how often the GC manager checks for
	// idle backends, not the idle timeout (which is controlled by MaxIdleDuration).
	GCIdleCheckInterval toml.Duration `toml:"gc-idle-check-interval"`
	// GCChannelBufferSize buffer size for GC task channels (gcInactiveC and createC).
	// Default is 1024. When channel is full, requests are dropped to avoid blocking.
	GCChannelBufferSize int `toml:"gc-channel-buffer-size"`

	// BackendOptions extra backend options
	BackendOptions []BackendOption `toml:"-"`
	// ClientOptions extra client options
	ClientOptions []ClientOption `toml:"-"`
	// CodecOptions extra codec options
	CodecOptions []CodecOption `toml:"-"`
}

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

type HandleFunc[REQ, RESP MethodBasedMessage] func(context.Context, REQ, RESP, *Buffer) error

type HandlerOption[REQ, RESP MethodBasedMessage] func(*methodBasedServer[REQ, RESP])

type HeaderCodec interface {
	// Encode encode header into output buffer
	Encode(*RPCMessage, *buf.ByteBuf) (int, error)
	// Decode decode header from input buffer
	Decode(*RPCMessage, []byte) (int, error)
}

type Message interface {
	// SetID each message has a unique ID in a RPCClient Backend. If it is a message transmitted
	// in stream, the ID must be set to Stream.ID.
	SetID(uint64)
	// GetID returns ID of the message
	GetID() uint64
	// DebugString return debug string
	DebugString() string
	// Size size of message after marshal
	ProtoSize() int
	// MarshalTo marshal to target byte slice
	MarshalTo(data []byte) (int, error)
	// Unmarshal unmarshal from data
	Unmarshal(data []byte) error
}

type MessageCache interface {
	// Add FIFO add message to cache
	Add(value Message) error
	// Len returns message count in the cache
	Len() (int, error)
	// Pop pop the first message in the cache, return false means no message in cache
	Pop() (Message, bool, error)
	// Close close the cache
	Close()
}

type MessagePool[REQ, RESP MethodBasedMessage] interface {
	AcquireRequest() REQ
	ReleaseRequest(REQ)

	AcquireResponse() RESP
	ReleaseResponse(RESP)
}

type MethodBasedClient[REQ, RESP MethodBasedMessage] interface {
	// RegisterMethod register remote getter func, used to get remote address by method.
	RegisterMethod(method uint32, remoteGetter func(REQ) (string, error))
	// Send send request to remote, and wait for a response synchronously.
	Send(context.Context, REQ) (RESP, error)
	// AsyncSend async send request to remote.
	AsyncSend(context.Context, REQ) (*Future, error)
	// Close close the client
	Close() error
}

type MethodBasedMessage interface {
	Message
	// Reset reset message
	Reset()
	// Method message type
	Method() uint32
	// SetMethod set message type.
	SetMethod(uint32)
	// WrapError wrap error into message, and transport to remote endpoint.
	WrapError(error)
	// UnwrapError parse error from the message.
	UnwrapError() error
}

type MethodBasedServer[REQ, RESP MethodBasedMessage] interface {
	// Start start the txn server
	Start() error
	// Close the txn server
	Close() error
	// RegisterMethod register request handler func
	RegisterMethod(method uint32, handleFunc HandleFunc[REQ, RESP], async bool) MethodBasedServer[REQ, RESP]
	// Handle handle at local
	Handle(ctx context.Context, req REQ, buf *Buffer) RESP
}

type PayloadMessage interface {
	Message

	// GetPayloadField return the payload data
	GetPayloadField() []byte
	// SetPayloadField set the payload data
	SetPayloadField(data []byte)
}

type RPCClient interface {
	// Send send a request message to the corresponding server and return a Future to get the
	// response message.
	Send(ctx context.Context, backend string, request Message) (*Future, error)
	// NewStream create a stream used to asynchronous stream of sending and receiving messages.
	// If the underlying connection is reset during the duration of the stream, then the stream will
	// be closed. The context is used to control cancellation of stream creation; if the context
	// is cancelled during connection establishment or retry backoff, NewStream will return immediately.
	NewStream(ctx context.Context, backend string, lock bool) (Stream, error)
	// Ping is used to check if the remote service is available. The remote service will reply with
	// a pong when it receives the ping.
	Ping(ctx context.Context, backend string) error
	// Close close the client
	Close() error

	CloseBackend() error
}

type RPCMessage struct {
	// Ctx context
	Ctx    context.Context
	Cancel context.CancelFunc
	// Message raw rpc message
	Message Message
	// contains filtered or unexported fields
}

type RPCServer interface {
	// Start start listening and wait for client messages. After each client link is established,
	// a separate goroutine is assigned to handle the Read, and the Read-to message is handed over
	// to the Handler for processing.
	Start() error
	// Close close the rpc server
	Close() error
	// RegisterRequestHandler register the request handler. The request handler is processed in the
	// read goroutine of the current client connection. Sequence is the sequence of message received
	// by the current client connection. If error returned by handler, client connection will closed.
	// Handler can use the ClientSession to write response, both synchronous and asynchronous.
	RegisterRequestHandler(func(ctx context.Context, request RPCMessage, sequence uint64, cs ClientSession) error)
}

type RetryPolicy struct {
	// MaxRetries is the maximum number of retry attempts.
	// 0 means unlimited (rely on context timeout), which is the default behavior.
	MaxRetries int
	// InitialBackoff is the initial backoff duration before the first retry.
	InitialBackoff time.Duration
	// MaxBackoff is the maximum backoff duration.
	MaxBackoff time.Duration
	// Multiplier is the factor by which backoff increases after each retry.
	Multiplier float64
	// JitterFraction adds randomness to backoff (0.2 means ±20%).
	JitterFraction float64
}

type ServerOption func(*server)

type StatusCategory int

type Stream interface {
	// ID returns the stream ID. All messages transmitted on the current stream need to use the
	// stream ID as the message ID
	ID() uint64
	// Send send message to stream
	Send(ctx context.Context, request Message) error
	// Receive returns a channel to read stream message from server. If nil is received, the receive
	// loop needs to exit. In any case, Stream.Close needs to be called.
	Receive() (chan Message, error)
	// Close close the stream. If closeConn is true, the underlying connection will be closed.
	Close(closeConn bool) error
}