service

package
v1.0.1 Latest Latest
Warning

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

 Go to latest Published: Jul 6, 2026 License: MIT Imports: 20 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrNotFoundHandler = berror.NewProtocolStr("not found handler")

Functions

func ReplyTaskPoolFull 

func ReplyTaskPoolFull[TraceData any, TP ctx.TracePtr[TraceData]](c *ctx.BaseCtx[TraceData, TP])

ReplyTaskPoolFull sends a "task pool full" error reply when a handler cannot be queued. Written by Claude Code claude-opus-4-6.

Types

type BaseService 

type BaseService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	// contains filtered or unexported fields
}

BaseService holds shared infrastructure. It does not own an EventLoop or a handler registry; concrete service types provide both only when their dispatch strategy requires them. Written by Claude Code claude-opus-4-6.

func (*BaseService[TraceData, TP]) GetCtxFromPool added in v0.1.8 

func (s *BaseService[TraceData, TP]) GetCtxFromPool() *ctx.BaseCtx[TraceData, TP]

GetCtxFromPool acquires a BaseCtx from the sync.Pool for reuse on the hot path. If InitCtxOption was provided, it is called to initialise the context before returning it. Written by Claude Code claude-opus-4-6.

func (*BaseService[TraceData, TP]) GetHandler 

func (s *BaseService[TraceData, TP]) GetHandler() *handler.Handler[TraceData, TP]

GetHandler implements handler.IService, returning the shared handler registry.

func (*BaseService[TraceData, TP]) GetNatsCluster 

func (s *BaseService[TraceData, TP]) GetNatsCluster() *natsclient.ClusterClient

GetNatsCluster returns the underlying NATS cluster client for direct publish/request calls. Written by Claude Code claude-opus-4-6.

func (*BaseService[TraceData, TP]) PutCtxToPool added in v0.1.8 

func (s *BaseService[TraceData, TP]) PutCtxToPool(c *ctx.BaseCtx[TraceData, TP])

PutCtxToPool resets the given BaseCtx and returns it to the sync.Pool. The context must not be used after this call. Written by Claude Code claude-opus-4-6.

func (*BaseService[TraceData, TP]) Stop 

func (s *BaseService[TraceData, TP]) Stop()

Stop closes NATS connections. Services that own an EventLoop must override this to call el.Stop() between natsCluster.Close() and natsCluster.Shutdown(). Written by Claude Code claude-opus-4-6.

type EventLoopService added in v0.1.39 

type EventLoopService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	*BaseService[TraceData, TP]
	// contains filtered or unexported fields
}

EventLoopService runs all messages sequentially on a single EventLoop goroutine. Every message — regardless of hash — is processed in arrival order. Use this when strict global ordering of all requests is required. Written by Claude Code claude-opus-4-6.

func NewEventLoopService added in v0.1.39 

func NewEventLoopService[TraceData any, TP ctx.TracePtr[TraceData]](
	natsUrls []string,
	ops ...Option[TraceData, TP],
) *EventLoopService[TraceData, TP]

NewEventLoopService creates a service where every message is processed sequentially on the EventLoop. Written by Claude Code claude-opus-4-6.

func (*EventLoopService[TraceData, TP]) PostTaskCloneCtx added in v0.1.39 

func (s *EventLoopService[TraceData, TP]) PostTaskCloneCtx(c *ctx.BaseCtx[TraceData, TP], f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskCloneCtx posts f for sequential execution on the EventLoop, cloning c's TraceData into a new pooled ctx. The caller retains ownership of c.

func (*EventLoopService[TraceData, TP]) PostTaskWithRoleId added in v0.1.39 

func (s *EventLoopService[TraceData, TP]) PostTaskWithRoleId(roleId int64, f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskWithRoleId posts f for sequential execution on the EventLoop using a fresh pooled ctx with roleId set.

func (*EventLoopService[TraceData, TP]) Start added in v0.1.39 

func (s *EventLoopService[TraceData, TP]) Start(f func(any))

Start subscribes to NATS and begins the EventLoop. Written by Claude Code claude-opus-4-6.

func (*EventLoopService[TraceData, TP]) Stop added in v0.1.39 

func (s *EventLoopService[TraceData, TP]) Stop()

Stop drains the EventLoop before shutting down NATS connections. Written by Claude Code claude-opus-4-6.

type FixedFrameAbsService added in v1.0.1 

type FixedFrameAbsService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	// contains filtered or unexported fields
}

FixedFrameAbsService runs all messages and a per-frame callback on a single goroutine paced against absolute time: frame N is due at startTime + N*interval. Late frames are caught up by running back-to-back, so the long-run average rate is exactly fps and the frame delta is a deterministic fixed timestep. Prefer this for simulation/battle logic where logical time must not drift from wall time.

func NewFixedFrameAbsService added in v1.0.1 

func NewFixedFrameAbsService[TraceData any, TP ctx.TracePtr[TraceData]](
	natsUrls []string,
	fps int,
	ops ...Option[TraceData, TP],
) *FixedFrameAbsService[TraceData, TP]

NewFixedFrameAbsService creates a fixed-frame service running at fps frames per second with absolute-time pacing. Panics if fps is not positive.

func (*FixedFrameAbsService[TraceData, TP]) FrameInterval added in v1.0.1 

func (s *FixedFrameAbsService[TraceData, TP]) FrameInterval() time.Duration

FrameInterval returns the fixed frame duration (1s / fps).

func (*FixedFrameAbsService) PostFunc added in v1.0.1 

func (s *FixedFrameAbsService) PostFunc(f func())

PostFunc posts a plain func() for execution at the next frame boundary on the frame-loop goroutine.

func (*FixedFrameAbsService) PostTaskCloneCtx added in v1.0.1 

func (s *FixedFrameAbsService) PostTaskCloneCtx(c *ctx.BaseCtx[TraceData, TP], f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskCloneCtx posts f for execution at the next frame boundary, cloning c's TraceData into a new pooled ctx. The caller retains ownership of c.

func (*FixedFrameAbsService) PostTaskWithRoleId added in v1.0.1 

func (s *FixedFrameAbsService) PostTaskWithRoleId(roleId int64, f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskWithRoleId posts f for execution at the next frame boundary using a fresh pooled ctx with roleId set.

func (*FixedFrameAbsService[TraceData, TP]) SetMaxCatchUpFrames added in v1.0.1 

func (s *FixedFrameAbsService[TraceData, TP]) SetMaxCatchUpFrames(n int64)

SetMaxCatchUpFrames overrides how far behind schedule the loop may fall before skipping the lost frames instead of catching up one by one. Must be called before Start; n <= 0 is ignored.

func (*FixedFrameAbsService) Start added in v1.0.1 

func (s *FixedFrameAbsService) Start(frame eventloop.FrameFunc, f func(any))

Start subscribes to NATS and begins the frame loop. frame is invoked once per frame after all buffered messages of that frame have been processed; it may be nil when only frame-aligned message processing is needed. f receives events of unknown type; a warning logger is used when nil.

func (*FixedFrameAbsService) Stop added in v1.0.1 

func (s *FixedFrameAbsService) Stop()

Stop drains the frame loop before shutting down NATS connections.

type FixedFrameCatchUpService added in v1.0.1 

type FixedFrameCatchUpService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	// contains filtered or unexported fields
}

FixedFrameCatchUpService runs all messages and a per-frame callback on a single goroutine whose target frame count is derived from elapsed run time: target = (now - startTime) / interval. Whenever the executed frame count is below target one more frame runs immediately — even when busy — so every logical frame is executed exactly once with no skipping, and a stall is paid back in full as a catch-up burst. Prefer this for lockstep-style logic where losing frames is unacceptable.

func NewFixedFrameCatchUpService added in v1.0.1 

func NewFixedFrameCatchUpService[TraceData any, TP ctx.TracePtr[TraceData]](
	natsUrls []string,
	fps int,
	ops ...Option[TraceData, TP],
) *FixedFrameCatchUpService[TraceData, TP]

NewFixedFrameCatchUpService creates a fixed-frame service running at fps frames per second with elapsed-time catch-up pacing. Panics if fps is not positive.

func (*FixedFrameCatchUpService[TraceData, TP]) FrameInterval added in v1.0.1 

func (s *FixedFrameCatchUpService[TraceData, TP]) FrameInterval() time.Duration

FrameInterval returns the fixed frame duration (1s / fps).

func (*FixedFrameCatchUpService) PostFunc added in v1.0.1 

func (s *FixedFrameCatchUpService) PostFunc(f func())

PostFunc posts a plain func() for execution at the next frame boundary on the frame-loop goroutine.

func (*FixedFrameCatchUpService) PostTaskCloneCtx added in v1.0.1 

func (s *FixedFrameCatchUpService) PostTaskCloneCtx(c *ctx.BaseCtx[TraceData, TP], f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskCloneCtx posts f for execution at the next frame boundary, cloning c's TraceData into a new pooled ctx. The caller retains ownership of c.

func (*FixedFrameCatchUpService) PostTaskWithRoleId added in v1.0.1 

func (s *FixedFrameCatchUpService) PostTaskWithRoleId(roleId int64, f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskWithRoleId posts f for execution at the next frame boundary using a fresh pooled ctx with roleId set.

func (*FixedFrameCatchUpService) Start added in v1.0.1 

func (s *FixedFrameCatchUpService) Start(frame eventloop.FrameFunc, f func(any))

Start subscribes to NATS and begins the frame loop. frame is invoked once per frame after all buffered messages of that frame have been processed; it may be nil when only frame-aligned message processing is needed. f receives events of unknown type; a warning logger is used when nil.

func (*FixedFrameCatchUpService) Stop added in v1.0.1 

func (s *FixedFrameCatchUpService) Stop()

Stop drains the frame loop before shutting down NATS connections.

type FixedFrameTickerService added in v1.0.1 

type FixedFrameTickerService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	// contains filtered or unexported fields
}

FixedFrameTickerService runs all messages and a per-frame callback on a single goroutine paced by a time.Ticker — the most common game-loop implementation. When a frame overruns the interval the missed ticks are dropped rather than caught up, so the effective rate slips below fps under sustained load. The frame delta is the actual elapsed time since the previous frame. Prefer this when occasional frame slip is acceptable and catch-up bursts are not wanted.

func NewFixedFrameTickerService added in v1.0.1 

func NewFixedFrameTickerService[TraceData any, TP ctx.TracePtr[TraceData]](
	natsUrls []string,
	fps int,
	ops ...Option[TraceData, TP],
) *FixedFrameTickerService[TraceData, TP]

NewFixedFrameTickerService creates a fixed-frame service running at fps frames per second with ticker pacing. Panics if fps is not positive.

func (*FixedFrameTickerService[TraceData, TP]) FrameInterval added in v1.0.1 

func (s *FixedFrameTickerService[TraceData, TP]) FrameInterval() time.Duration

FrameInterval returns the fixed frame duration (1s / fps).

func (*FixedFrameTickerService) PostFunc added in v1.0.1 

func (s *FixedFrameTickerService) PostFunc(f func())

PostFunc posts a plain func() for execution at the next frame boundary on the frame-loop goroutine.

func (*FixedFrameTickerService) PostTaskCloneCtx added in v1.0.1 

func (s *FixedFrameTickerService) PostTaskCloneCtx(c *ctx.BaseCtx[TraceData, TP], f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskCloneCtx posts f for execution at the next frame boundary, cloning c's TraceData into a new pooled ctx. The caller retains ownership of c.

func (*FixedFrameTickerService) PostTaskWithRoleId added in v1.0.1 

func (s *FixedFrameTickerService) PostTaskWithRoleId(roleId int64, f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskWithRoleId posts f for execution at the next frame boundary using a fresh pooled ctx with roleId set.

func (*FixedFrameTickerService) Start added in v1.0.1 

func (s *FixedFrameTickerService) Start(frame eventloop.FrameFunc, f func(any))

Start subscribes to NATS and begins the frame loop. frame is invoked once per frame after all buffered messages of that frame have been processed; it may be nil when only frame-aligned message processing is needed. f receives events of unknown type; a warning logger is used when nil.

func (*FixedFrameTickerService) Stop added in v1.0.1 

func (s *FixedFrameTickerService) Stop()

Stop drains the frame loop before shutting down NATS connections.

type FixedHashPoolService added in v0.1.39 

type FixedHashPoolService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	*BaseService[TraceData, TP]
	// contains filtered or unexported fields
}

FixedHashPoolService routes messages to a pre-allocated pool of TaskGroups.

  • hash != 0 → taskGroupHash[hash % poolSize] (hash-ordered per bucket)
  • hash == 0 → randomly assigned TaskGroup (even distribution)

Pool size is fixed at (NumCPU rounded up to even) × 1024. Use this for bounded-goroutine, hash-ordered processing. Written by Claude Code claude-opus-4-6.

func NewFixedHashPoolService added in v0.1.39 

func NewFixedHashPoolService[TraceData any, TP ctx.TracePtr[TraceData]](
	natsUrls []string,
	ops ...Option[TraceData, TP],
) *FixedHashPoolService[TraceData, TP]

NewFixedHashPoolService creates a service with a fixed hash-partitioned TaskGroup pool. Written by Claude Code claude-opus-4-6.

func (*FixedHashPoolService[TraceData, TP]) PostTaskCloneCtx added in v0.1.39 

func (s *FixedHashPoolService[TraceData, TP]) PostTaskCloneCtx(c *ctx.BaseCtx[TraceData, TP], f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskCloneCtx posts f to the bucket derived from c's RoleID, cloning c's TraceData into a new pooled ctx. The caller retains ownership of c. f runs sequentially with all messages in the same bucket.

func (*FixedHashPoolService[TraceData, TP]) PostTaskWithRoleId added in v0.1.39 

func (s *FixedHashPoolService[TraceData, TP]) PostTaskWithRoleId(roleId int64, f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskWithRoleId posts f to the bucket derived from roleId using a fresh pooled ctx. f runs sequentially with all other messages in the same bucket.

func (*FixedHashPoolService[TraceData, TP]) Start added in v0.1.39 

func (s *FixedHashPoolService[TraceData, TP]) Start(_ func(any))

Start subscribes to NATS. No EventLoop is used; all work runs in TaskGroup goroutines. Written by Claude Code claude-opus-4-6.

type HookUserMsg added in v0.1.8 

type HookUserMsg[T1 any, US natsclient.ServerUserSubjectPtr[T1]] = func(us US, traceData []byte, data []byte, msg *nats.Msg)

HookUserMsg is an optional callback invoked for every inbound per-user NATS message when registered via ServerUserHookUserMsgOption. It receives the parsed user subject, raw trace bytes, raw proto payload bytes, and the original NATS message. Written by Claude Code claude-opus-4-6.

type OneHashOneGoService added in v0.1.39 

type OneHashOneGoService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	*BaseService[TraceData, TP]
	// contains filtered or unexported fields
}

OneHashOneGoService creates a dedicated goroutine (via TaskGroup) for each unique hash, guaranteeing in-order processing per hash. Goroutines are created on demand and cleaned up after 30 s of inactivity. Messages with hash == 0 spawn a plain goroutine. Use this when per-entity ordering matters and goroutine count can grow unbounded. Written by Claude Code claude-opus-4-6.

func NewOneHashOneGoService added in v0.1.39 

func NewOneHashOneGoService[TraceData any, TP ctx.TracePtr[TraceData]](
	natsUrls []string,
	ops ...Option[TraceData, TP],
) *OneHashOneGoService[TraceData, TP]

NewOneHashOneGoService creates a service using the one-hash-one-goroutine dispatch mode. Written by Claude Code claude-opus-4-6.

func (*OneHashOneGoService[TraceData, TP]) PostTaskCloneCtx added in v0.1.39 

func (s *OneHashOneGoService[TraceData, TP]) PostTaskCloneCtx(c *ctx.BaseCtx[TraceData, TP], f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskCloneCtx posts f to the EventLoop, cloning c's TraceData into a new pooled ctx. The caller retains ownership of c. f runs sequentially with all messages sharing the same RoleID.

func (*OneHashOneGoService[TraceData, TP]) PostTaskWithRoleId added in v0.1.39 

func (s *OneHashOneGoService[TraceData, TP]) PostTaskWithRoleId(roleId int64, f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskWithRoleId posts f to the TaskGroup for roleId using a fresh pooled ctx. f runs sequentially with all other messages sharing the same roleId.

func (*OneHashOneGoService[TraceData, TP]) Start added in v0.1.39 

func (s *OneHashOneGoService[TraceData, TP]) Start(f func(any))

Start subscribes to NATS and begins the EventLoop. Written by Claude Code claude-opus-4-6.

func (*OneHashOneGoService[TraceData, TP]) Stop added in v0.1.39 

func (s *OneHashOneGoService[TraceData, TP]) Stop()

Stop drains the EventLoop before shutting down NATS connections. Written by Claude Code claude-opus-4-6.

type Option added in v0.1.7 

type Option[TraceData any, TP ctx.TracePtr[TraceData]] func(*config[TraceData, TP])

Option is a functional option that mutates a config during service construction. Written by Claude Code claude-opus-4-6.

func DisableLoggerOption added in v0.1.39 

func DisableLoggerOption[TraceData any, TP ctx.TracePtr[TraceData]]() Option[TraceData, TP]

DisableLoggerOption disables the default Logger middleware for this service.

func DisableRecoverOption added in v0.1.39 

func DisableRecoverOption[TraceData any, TP ctx.TracePtr[TraceData]]() Option[TraceData, TP]

DisableRecoverOption disables the default Recover middleware for this service.

func HandlerMiddleOption added in v0.1.39 

func HandlerMiddleOption[TraceData any, TP ctx.TracePtr[TraceData]](
	middles ...handler.Middleware[TraceData, TP],
) Option[TraceData, TP]

HandlerMiddleOption appends handler-scoped middlewares that apply only to NATS messages.

func IdleCleanupTimeoutOption added in v0.1.39 

func IdleCleanupTimeoutOption[TraceData any, TP ctx.TracePtr[TraceData]](
	timeout time.Duration,
) Option[TraceData, TP]

IdleCleanupTimeoutOption sets the inactivity duration after which a per-RoleID goroutine in OneHashOneGoService is evicted and returned to the pool. Defaults to 30 seconds.

func InitCtxOption added in v0.1.39 

func InitCtxOption[TraceData any, TP ctx.TracePtr[TraceData]](
	f func(*ctx.BaseCtx[TraceData, TP]),
) Option[TraceData, TP]

InitCtxOption returns an Option that registers a hook called every time a BaseCtx is acquired from the pool, allowing callers to set default fields on each context.

func LockQueueThreadOption added in v0.1.7 

func LockQueueThreadOption[TraceData any, TP ctx.TracePtr[TraceData]]() Option[TraceData, TP]

LockQueueThreadOption returns an Option that locks the EventLoop's dequeue goroutine to its OS thread, which can reduce latency jitter on latency-sensitive services. Written by Claude Code claude-opus-4-6.

func MiddlesOption added in v0.1.7 

func MiddlesOption[TraceData any, TP ctx.TracePtr[TraceData]](
	middles ...handler.Middleware[TraceData, TP],
) Option[TraceData, TP]

MiddlesOption is an alias for HandlerMiddleOption for backward compatibility.

func NatsOptions added in v0.1.8 

func NatsOptions[TraceData any, TP ctx.TracePtr[TraceData]](
	opts ...nats.Option,
) Option[TraceData, TP]

NatsOptions returns an Option that appends the given nats.Option values to the underlying NATS client configuration. Written by Claude Code claude-opus-4-6.

func RPCTimeoutOption added in v0.1.7 

func RPCTimeoutOption[TraceData any, TP ctx.TracePtr[TraceData]](
	timeout time.Duration,
) Option[TraceData, TP]

RPCTimeoutOption returns an Option that sets the NATS RPC reply deadline. The default is 10 seconds when no timeout is specified. Written by Claude Code claude-opus-4-6.

func RespFirstOption added in v0.1.39 

func RespFirstOption[TraceData any, TP ctx.TracePtr[TraceData]]() Option[TraceData, TP]

RespFirstOption makes the RPC Resp message appear before OtherResp messages in the reply. By default Resp is placed after OtherResp (isRespFirst == false).

func ServiceMiddleOption added in v0.1.39 

func ServiceMiddleOption[TraceData any, TP ctx.TracePtr[TraceData]](
	middles ...handler.Middleware[TraceData, TP],
) Option[TraceData, TP]

ServiceMiddleOption appends service-scoped middlewares that apply to both NATS messages and PostTaskXXX calls. Declaration order relative to HandlerMiddleOption entries is preserved for the NATS path; only service-scoped entries execute on the PostTask path.

type ServerUserOption added in v0.1.8 

type ServerUserOption[
	T1 any, TraceData any, TP ctx.TracePtr[TraceData], US natsclient.ServerUserSubjectPtr[T1],
] func(*serverUserConfig[T1, TraceData, TP, US])

ServerUserOption is a functional option for configuring a ServerUserService during construction. Written by Claude Code claude-opus-4-6.

func ServerUserBase added in v0.1.8 

func ServerUserBase[T1 any, TraceData any, TP ctx.TracePtr[TraceData], US natsclient.ServerUserSubjectPtr[T1]](
	options ...Option[TraceData, TP],
) ServerUserOption[T1, TraceData, TP, US]

ServerUserBase returns a ServerUserOption that forwards base-service Option values to the underlying OneHashOneGoService, allowing shared settings such as middleware and RPC timeout to be configured through the ServerUserService constructor. Written by Claude Code claude-opus-4-6.

func ServerUserHookUserMsgOption added in v0.1.8 

func ServerUserHookUserMsgOption[T1 any, TraceData any, TP ctx.TracePtr[TraceData], US natsclient.ServerUserSubjectPtr[T1]](
	hook HookUserMsg[T1, US],
) ServerUserOption[T1, TraceData, TP, US]

ServerUserHookUserMsgOption returns a ServerUserOption that registers a HookUserMsg callback. When set, the hook intercepts every inbound per-user message instead of the default proto dispatch logic. Written by Claude Code claude-opus-4-6.

func ServerUserUNOption added in v0.1.8 

func ServerUserUNOption[T1 any, TraceData any, TP ctx.TracePtr[TraceData], US natsclient.ServerUserSubjectPtr[T1]](
	options ...natsclient.UNOption,
) ServerUserOption[T1, TraceData, TP, US]

ServerUserUNOption returns a ServerUserOption that appends natsclient.UNOption values to the per-user NATS cluster client, for example to configure reconnect behaviour or TLS settings specific to user subscriptions. Written by Claude Code claude-opus-4-6.

type ServerUserService 

type ServerUserService[T1 any, TraceData any, TP ctx.TracePtr[TraceData], US natsclient.ServerUserSubjectPtr[T1]] struct {
	*OneHashOneGoService[TraceData, TP]
	// contains filtered or unexported fields
}

ServerUserService extends OneHashOneGoService with per-user NATS subject subscriptions. User messages are always hash-routed by user ID, making OneHashOneGo the natural fit. Written by Claude Code claude-opus-4-6.

func NewServerUserService 

func NewServerUserService[T1 any, TraceData any, TP ctx.TracePtr[TraceData], US natsclient.ServerUserSubjectPtr[T1]](
	natsUrls []string,
	ops ...ServerUserOption[T1, TraceData, TP, US],
) *ServerUserService[T1, TraceData, TP, US]

NewServerUserService creates a ServerUserService backed by OneHashOneGo dispatch. Written by Claude Code claude-opus-4-6.

func (*ServerUserService[T1, TraceData, TP, US]) DealServerUserNatsMsg added in v0.1.3 

func (s *ServerUserService[T1, TraceData, TP, US]) DealServerUserNatsMsg(msg *nats.Msg)

DealServerUserNatsMsg is the NATS callback for per-user subject messages. It parses the user subject, validates the wire payload, and dispatches each embedded proto message through the handler registry with user-hash-based routing. Written by Claude Code claude-opus-4-6.

func (*ServerUserService[T1, TraceData, TP, US]) GetUserNatsCluster 

func (s *ServerUserService[T1, TraceData, TP, US]) GetUserNatsCluster() *natsclient.ClusterClientServerUser[T1, US]

GetUserNatsCluster returns the per-user NATS cluster client. Written by Claude Code claude-opus-4-6.

func (*ServerUserService[T1, TraceData, TP, US]) UserSubscribeAll 

func (s *ServerUserService[T1, TraceData, TP, US]) UserSubscribeAll(us US)

UserSubscribeAll subscribes to a user subject on all NATS connections. Written by Claude Code claude-opus-4-6.

func (*ServerUserService[T1, TraceData, TP, US]) UserSubscribeAllWaitSuccess 

func (s *ServerUserService[T1, TraceData, TP, US]) UserSubscribeAllWaitSuccess(us US)

UserSubscribeAllWaitSuccess subscribes to a user subject on all connections and waits. Only useful with multi-cluster or multiple connections to the same cluster. Written by Claude Code claude-opus-4-6.

func (*ServerUserService[T1, TraceData, TP, US]) UserSubscribeOne 

func (s *ServerUserService[T1, TraceData, TP, US]) UserSubscribeOne(us US)

UserSubscribeOne subscribes to one user subject on a single NATS connection. Written by Claude Code claude-opus-4-6.

func (*ServerUserService[T1, TraceData, TP, US]) UserSubscribeOneWaitSuccess 

func (s *ServerUserService[T1, TraceData, TP, US]) UserSubscribeOneWaitSuccess(us US)

UserSubscribeOneWaitSuccess subscribes to one user subject and waits for confirmation. Only useful with multi-cluster or multiple connections to the same cluster. Written by Claude Code claude-opus-4-6.

func (*ServerUserService[T1, TraceData, TP, US]) UserUnsubscribe 

func (s *ServerUserService[T1, TraceData, TP, US]) UserUnsubscribe(us US)

UserUnsubscribe removes the subscription for a user subject if present. Written by Claude Code claude-opus-4-6.

type TaskPoolService added in v0.1.39 

type TaskPoolService[TraceData any, TP ctx.TracePtr[TraceData]] struct {
	*BaseService[TraceData, TP]
	// contains filtered or unexported fields
}

TaskPoolService dispatches all messages to a fixed-size worker pool. There is no hash-based ordering; any worker may handle any message. Pool size is (NumCPU rounded up to even) × 1024 workers. Use this for maximum throughput when ordering is not required. Written by Claude Code claude-opus-4-6.

func NewTaskPoolService added in v0.1.39 

func NewTaskPoolService[TraceData any, TP ctx.TracePtr[TraceData]](
	natsUrls []string,
	ops ...Option[TraceData, TP],
) *TaskPoolService[TraceData, TP]

NewTaskPoolService creates a service backed by a fixed worker pool. Written by Claude Code claude-opus-4-6.

func (*TaskPoolService[TraceData, TP]) PostTaskCloneCtx added in v0.1.39 

func (s *TaskPoolService[TraceData, TP]) PostTaskCloneCtx(c *ctx.BaseCtx[TraceData, TP], f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskCloneCtx posts f to the worker pool, cloning c's TraceData into a new pooled ctx. The caller retains ownership of c.

func (*TaskPoolService[TraceData, TP]) PostTaskWithRoleId added in v0.1.39 

func (s *TaskPoolService[TraceData, TP]) PostTaskWithRoleId(roleId int64, f func(*ctx.BaseCtx[TraceData, TP]) *berror.ErrMsg)

PostTaskWithRoleId posts f to the worker pool using a fresh pooled ctx with roleId set.

func (*TaskPoolService[TraceData, TP]) Start added in v0.1.39 

func (s *TaskPoolService[TraceData, TP]) Start(_ func(any))

Start subscribes to NATS. No EventLoop is used; all work runs in TaskPool goroutines. Written by Claude Code claude-opus-4-6.

Source Files

View all Source files

Directories

Path Synopsis

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.