csms

package
v0.2.0 Latest Latest
Warning

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

 Go to latest Published: Jul 23, 2026 License: MIT Imports: 17 Imported by: 0

Documentation

Overview

Package csms provides an OCPP-J Central System Management System server.

Server owns WebSocket connections and session lifecycles. Router dispatches inbound CALL messages to application-provided typed handlers, while Call sends typed requests from the CSMS to a connected Charging Station.

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrInvalidHandlerRegistration = errors.New("invalid OCPP handler registration")
	ErrHandlerAlreadyRegistered   = errors.New("OCPP handler is already registered")
)
View Source 
var (
	ErrAuthenticationFailed      = errors.New("authentication failed")
	ErrAuthenticationUnavailable = errors.New("authentication service unavailable")
)
View Source 
var (
	ErrInvalidConfiguration = errors.New("invalid CSMS configuration")
	ErrUniqueIDGeneration   = errors.New("OCPP unique ID generation failed")
)
View Source 
var (
	ErrSessionClosed       = errors.New("OCPP session is closed")
	ErrSessionReplaced     = errors.New("OCPP session was replaced")
	ErrServerShutdown      = errors.New("OCPP server is shutting down")
	ErrPongTimeout         = errors.New("WebSocket pong timeout")
	ErrIdleTimeout         = errors.New("OCPP session idle timeout")
	ErrTooManyPendingCalls = errors.New("too many pending OCPP calls")
	ErrDuplicateUniqueID   = errors.New("duplicate OCPP unique ID")
	ErrHandlerQueueFull    = errors.New("OCPP handler queue is full")
)

Functions

func Call 

func Call[Request protocol.Payload, Confirmation protocol.Payload](
	ctx context.Context,
	session *Session,
	request Request,
) (Confirmation, error)

Call sends a generated request to a Charging Station and waits for the matching generated confirmation or CALLERROR.

func Handle 

func Handle[Request protocol.Payload, Confirmation protocol.Payload](
	router *Router,
	handler TypedHandler[Request, Confirmation],
) error

Handle registers a type-safe OCPP handler. Request and Confirmation must be generated non-pointer payload types for the same action and version.

func HandleAfter added in v0.2.0 

func HandleAfter[Request protocol.Payload, Confirmation protocol.Payload](
	router *Router,
	handler TypedAfterHandler[Request, Confirmation],
) error

HandleAfter registers a type-safe post-response callback. Multiple callbacks for the same action are allowed and run in registration order.

func HandleSend 

func HandleSend[Request protocol.Payload](router *Router, handler TypedSendHandler[Request]) error

HandleSend registers an OCPP 2.1 SEND handler. SEND is unconfirmed, so an error returned by the handler is not written back to the Charging Station.

func ValidateIdentity 

func ValidateIdentity(identity string) error

ValidateIdentity accepts RFC 3986 unreserved characters and limits an OCPP identity to 255 bytes. Applications may replace it in SecurityConfig.

Types

type AfterHandler added in v0.2.0 

type AfterHandler func(context.Context, *Session, json.RawMessage, any) error

AfterHandler runs after a CALLRESULT has been written successfully. The request is the original CALL payload and confirmation is the value returned by the main handler. Its error is diagnostic only because the peer has already received the successful response.

type AuthenticationRequest 

type AuthenticationRequest struct {
	Identity   string
	Version    protocol.Version
	RemoteAddr string
	TLS        *tls.ConnectionState
	Basic      *BasicCredentials
}

type Authenticator 

type Authenticator interface {
	Authenticate(context.Context, AuthenticationRequest) (Principal, error)
}

type AuthenticatorFunc 

type AuthenticatorFunc func(context.Context, AuthenticationRequest) (Principal, error)

func (AuthenticatorFunc) Authenticate 

func (function AuthenticatorFunc) Authenticate(ctx context.Context, request AuthenticationRequest) (Principal, error)

type BasicCredentials 

type BasicCredentials struct {
	Username string
	Password []byte
}

type CallError 

type CallError struct {
	Code        ErrorCode
	Description string
	Details     any
}

func (*CallError) Error 

func (e *CallError) Error() string

type Config 

type Config struct {
	Router                *Router
	Versions              []protocol.Version
	ReadLimit             int64
	HandshakeTimeout      time.Duration
	CallTimeout           time.Duration
	MaxPendingCalls       int
	MaxConcurrentHandlers int
	// MaxQueuedHandlers bounds how many inbound CALL/SEND messages may be
	// admitted (running or waiting for a free MaxConcurrentHandlers slot)
	// at once, per session. Default: 4x MaxConcurrentHandlers. Once full, a
	// further inbound CALL gets an immediate CALLERROR (GenericError) and a
	// further inbound SEND is logged and dropped, instead of either
	// blocking the read loop (which would also stall reading this
	// session's own outbound Call responses) or admitting an unbounded
	// number of goroutines waiting for a slot.
	MaxQueuedHandlers  int
	UniqueIDGenerator  UniqueIDGenerator
	OnDuplicateSession DuplicateSessionHandler
	WriteTimeout       time.Duration
	PingInterval       time.Duration
	PongTimeout        time.Duration
	IdleTimeout        time.Duration
	CheckOrigin        func(*http.Request) bool
	OnConnect          func(*Session)
	OnDisconnect       func(*Session, error)
	Logger             Logger
	Metrics            Metrics
	Security           SecurityConfig
}

Config controls a CSMS Server. Construct it with keyed fields: new optional fields may be added in backward-compatible releases. Zero values select the documented default except IdleTimeout, where zero disables idle detection. Negative durations and limits are rejected.

type DuplicateSessionAttempt 

type DuplicateSessionAttempt struct {
	Identity string
	Existing SessionInfo
	Request  *http.Request
}

type DuplicateSessionDecision 

type DuplicateSessionDecision uint8
const (
	RejectNewSession DuplicateSessionDecision = iota + 1
	ReplaceExistingSession
)

type DuplicateSessionHandler 

type DuplicateSessionHandler func(context.Context, DuplicateSessionAttempt) (DuplicateSessionDecision, error)

type ErrorCode 

type ErrorCode string
const (
	NotImplemented                ErrorCode = "NotImplemented"
	NotSupported                  ErrorCode = "NotSupported"
	FormationViolation            ErrorCode = "FormationViolation"
	FormatViolation               ErrorCode = "FormatViolation"
	InternalError                 ErrorCode = "InternalError"
	ProtocolError                 ErrorCode = "ProtocolError"
	SecurityError                 ErrorCode = "SecurityError"
	PropertyConstraintViolation   ErrorCode = "PropertyConstraintViolation"
	OccurenceConstraintViolation  ErrorCode = "OccurenceConstraintViolation"
	OccurrenceConstraintViolation ErrorCode = "OccurrenceConstraintViolation"
	TypeConstraintViolation       ErrorCode = "TypeConstraintViolation"
	MessageTypeNotSupported       ErrorCode = "MessageTypeNotSupported"
	RpcFrameworkError             ErrorCode = "RpcFrameworkError"
	GenericError                  ErrorCode = "GenericError"
)

type Handler 

type Handler func(context.Context, *Session, json.RawMessage) (any, error)

type HandshakeAttempt 

type HandshakeAttempt struct {
	Identity   string
	Version    protocol.Version
	RemoteAddr string
}

type HandshakeLimiter 

type HandshakeLimiter interface {
	Allow(context.Context, HandshakeAttempt) bool
}

type HandshakeLimiterFunc 

type HandshakeLimiterFunc func(context.Context, HandshakeAttempt) bool

func (HandshakeLimiterFunc) Allow 

func (function HandshakeLimiterFunc) Allow(ctx context.Context, attempt HandshakeAttempt) bool

type IPRateLimiter 

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

IPRateLimiter limits WebSocket handshake attempts per remote IP using fixed windows. It is safe for concurrent use.

func NewIPRateLimiter 

func NewIPRateLimiter(limit int, window time.Duration) (*IPRateLimiter, error)

func (*IPRateLimiter) Allow 

func (limiter *IPRateLimiter) Allow(_ context.Context, attempt HandshakeAttempt) bool

type IdentityValidator 

type IdentityValidator func(string) error

type LogEvent 

type LogEvent string
const (
	LogSessionConnected    LogEvent = "session.connected"
	LogSessionDisconnected LogEvent = "session.disconnected"
	LogCallReceived        LogEvent = "call.received"
	LogCallCompleted       LogEvent = "call.completed"
	LogCallRejected        LogEvent = "call.rejected"
	LogCallAfterFailed     LogEvent = "call.after_failed"
	LogSendReceived        LogEvent = "send.received"
	LogSendCompleted       LogEvent = "send.completed"
	LogSendDropped         LogEvent = "send.dropped"
)

type LogLevel 

type LogLevel uint8
const (
	LogDebug LogLevel = iota + 1
	LogInfo
	LogWarn
	LogError
)

type LogRecord 

type LogRecord struct {
	Level       LogLevel
	Event       LogEvent
	Identity    string
	Version     protocol.Version
	MessageType protocol.MessageTypeID
	MessageID   string
	Action      string
	ErrorCode   ErrorCode
	Reason      string
}

LogRecord contains protocol metadata only. Payloads, credentials, certificates, idTokens and handler error text are intentionally omitted.

type Logger 

type Logger interface {
	Log(context.Context, LogRecord)
}

type LoggerFunc 

type LoggerFunc func(context.Context, LogRecord)

func (LoggerFunc) Log 

func (f LoggerFunc) Log(ctx context.Context, record LogRecord)

type MetricEvent added in v0.2.0 

type MetricEvent struct {
	Type MetricEventType
	// Identity is the charging station's identity string. It is NOT safe to
	// use directly as a label in a cardinality-sensitive metrics backend
	// (for example a Prometheus label) in a large deployment: every
	// distinct Identity creates a new time series, and a fleet of
	// thousands of charging stations produces thousands of series per
	// metric — a well-known Prometheus operational failure mode. This is a
	// deployment-scale judgment call the library cannot make correctly for
	// every user (fine for a 10-charger depot, wrong for a 50,000-charger
	// public network), so it is left entirely to the Metrics
	// implementation: use Identity for filtering/branching logic (e.g.
	// route a subset of chargers to more detailed stats) rather than
	// passing it straight through as a label, unless your deployment is
	// small enough that per-charger series are acceptable. See
	// examples/prometheus-hook for a wiring example that deliberately
	// excludes Identity from its labels.
	Identity    string
	Version     protocol.Version
	MessageType protocol.MessageTypeID
	Action      string
	Duration    time.Duration
	ErrorCode   ErrorCode
}

MetricEvent describes one measurable protocol event. Unlike LogRecord, it is not meant for human-readable diagnostics: it carries Duration for latency observation and has no free-form reason text, so implementations can use Type/Action/Version/ErrorCode directly as low-cardinality metric labels. Identity is deliberately not in that list — see its own comment.

type MetricEventType added in v0.2.0 

type MetricEventType uint8
const (
	// MetricSessionConnected is recorded as early as possible in the
	// connection lifecycle, before the read loop starts and before
	// Config.OnConnect runs. It is still not guaranteed to be the first
	// event Metrics.Record observes for a session: dispatch is async and
	// unordered (see the Metrics doc comment), so a Metrics implementation
	// must not assume Connected always precedes every other event for the
	// same identity.
	MetricSessionConnected MetricEventType = iota + 1
	// MetricSessionDisconnected carries Duration equal to the session's
	// total lifetime (time since connect).
	MetricSessionDisconnected
	MetricCallReceived
	// MetricCallCompleted carries Duration for validation, handler
	// execution, and the CALLRESULT write.
	MetricCallCompleted
	// MetricCallRejected carries Duration and ErrorCode.
	MetricCallRejected
	MetricSendReceived
	MetricSendCompleted
	MetricSendDropped
	// MetricOutboundCallRejected fires when Call is rejected before a CALL
	// frame is ever sent because MaxPendingCalls was reached. Duration is
	// always zero.
	MetricOutboundCallRejected
	// MetricOutboundCallSent fires once the CALL frame has been written.
	// Duration is always zero.
	MetricOutboundCallSent
	// MetricOutboundCallCompleted carries Duration measured from just after
	// the call was admitted into the pending-call table (i.e. it includes
	// CALL marshal/write time, not strictly the interval after
	// MetricOutboundCallSent) to a valid CALLRESULT.
	MetricOutboundCallCompleted
	// MetricOutboundCallFailed covers a remote CALLERROR, an undecodable or
	// invalid confirmation, a transport write failure unrelated to context
	// cancellation, and the session closing while the call was pending.
	// ErrorCode is set only for a remote CALLERROR. Note: this also fires
	// for every pending call during an ordinary graceful Server.Shutdown
	// (the session closes with ErrServerShutdown), which is expected and
	// not a regression.
	MetricOutboundCallFailed
	// MetricOutboundCallTimeout fires when the call's context deadline
	// expires, including when that deadline has already passed at the
	// moment the CALL frame is written. It cannot distinguish
	// Config.CallTimeout from a deadline the caller supplied on its own
	// context.
	MetricOutboundCallTimeout
	// MetricOutboundCallCanceled fires when the caller cancels its own
	// context while the call is pending (including if it was already
	// canceled at the moment the CALL frame is written). This is not a
	// failure. Caveat: context.Context only exposes context.Canceled, not
	// why a context was canceled, so if a caller ties Call's ctx to
	// Session.Context() (or another context whose cancellation is driven by
	// the session's own lifecycle rather than the caller's own decision),
	// the session closing for a reason unrelated to the call itself
	// (Server.Shutdown, an idle/pong timeout, being replaced) is also
	// reported here rather than as MetricOutboundCallFailed — this hook
	// cannot distinguish the two cases from ctx.Err() alone.
	MetricOutboundCallCanceled
)

type Metrics added in v0.2.0 

type Metrics interface {
	Record(context.Context, MetricEvent)
}

Metrics receives MetricEvent notifications for connection, CALL, and SEND lifecycle events. Implementations are expected to aggregate events (for example into counters and histograms) rather than perform I/O per event.

Record must be safe for concurrent use: a fixed worker pool shared by all sessions on a Server dispatches events, so Record may run concurrently with itself and carries no ordering guarantee relative to when events logically occurred. A slow or blocking Record implementation never delays the protocol server — recordMetric only attempts a non-blocking enqueue — but permanently blocking calls reduce the worker pool's effective capacity. If producers remain faster than the available workers and the bounded queue fills, further events are silently dropped rather than blocking.

The context passed to Record is always context.Background(), never a request- or session-scoped context: every MetricEvent can be produced at a point where the natural originating context (an inbound handler's context, an outbound Call's context, or the session's own context) is already canceled — most obviously for the very events that report a timeout, cancellation, or shutdown — and propagating it would let a defensively written Record implementation (one that no-ops when ctx.Err() != nil) silently drop exactly those events. A future tracing integration that needs span correlation will need its own mechanism rather than relying on this ctx.

type MetricsFunc added in v0.2.0 

type MetricsFunc func(context.Context, MetricEvent)

func (MetricsFunc) Record added in v0.2.0 

func (f MetricsFunc) Record(ctx context.Context, event MetricEvent)

type Middleware 

type Middleware func(protocol.Version, string, Handler) Handler

type Principal 

type Principal struct {
	ID         string
	Attributes map[string]string
}

type RemoteCallError 

type RemoteCallError struct {
	Code        ErrorCode
	Description string
	Details     json.RawMessage
}

RemoteCallError is a CALLERROR returned by a Charging Station.

func (*RemoteCallError) Error 

func (e *RemoteCallError) Error() string

type Router 

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

func NewRouter 

func NewRouter() *Router

func (*Router) Handle 

func (r *Router) Handle(version protocol.Version, action string, handler Handler) error

Handle registers a new CALL-answering handler for version and action. A registration is immutable: duplicate keys are rejected instead of silently replacing the existing handler. Only a CALL frame can reach a handler registered this way — use HandleSend to register a handler for SEND (fire-and-forget) frames instead.

func (*Router) HandleAfter added in v0.2.0 

func (r *Router) HandleAfter(version protocol.Version, action string, handler AfterHandler) error

HandleAfter appends a callback that runs after a successful CALLRESULT write. An action may have multiple callbacks; they run in registration order.

func (*Router) HandleSend added in v0.2.0 

func (r *Router) HandleSend(version protocol.Version, action string, handler Handler) error

HandleSend registers a new SEND (fire-and-forget) handler for version and action, following the same immutability rule as Handle. Only a SEND frame can reach a handler registered this way.

func (*Router) Use 

func (r *Router) Use(middleware Middleware)

Use appends middleware that is applied to every registered handler at lookup time. Middleware is executed in registration order.

type SecurityConfig 

type SecurityConfig struct {
	Profile                    SecurityProfile
	Authenticator              Authenticator
	ValidateIdentity           IdentityValidator
	AllowBasicUsernameMismatch bool
	MinTLSVersion              uint16
	HandshakeLimiter           HandshakeLimiter
	OnEvent                    SecurityEventHandler
}

type SecurityEvent 

type SecurityEvent struct {
	Type       SecurityEventType
	Identity   string
	Version    protocol.Version
	RemoteAddr string
	Reason     string
}

type SecurityEventHandler 

type SecurityEventHandler func(context.Context, SecurityEvent)

type SecurityEventType 

type SecurityEventType string
const (
	SecurityEventAuthenticationSucceeded SecurityEventType = "AuthenticationSucceeded"
	SecurityEventAuthenticationFailed    SecurityEventType = "AuthenticationFailed"
	SecurityEventHandshakeRateLimited    SecurityEventType = "HandshakeRateLimited"
	SecurityEventDuplicateRejected       SecurityEventType = "DuplicateSessionRejected"
	SecurityEventSessionReplaced         SecurityEventType = "SessionReplaced"
)

type SecurityProfile 

type SecurityProfile uint8
const (
	SecurityProfileUnsecured SecurityProfile = iota
	SecurityProfileBasicAuth
	SecurityProfileTLSBasicAuth
	SecurityProfileTLSClientCertificate
)

type Server 

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

func New 

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

func (*Server) Healthy added in v0.2.0 

func (s *Server) Healthy() bool

Healthy reports whether the server is still accepting new connections. It becomes false as soon as Shutdown is called.

func (*Server) ServeHTTP 

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP accepts /{chargePointIdentity} and negotiates an OCPP subprotocol.

func (*Server) Session 

func (s *Server) Session(identity string) (*Session, bool)

func (*Server) SessionCount 

func (s *Server) SessionCount() int

func (*Server) Sessions 

func (s *Server) Sessions() []*Session

func (*Server) Shutdown 

func (s *Server) Shutdown(ctx context.Context) error

Shutdown rejects new connections, closes active sessions and waits until they finish or ctx expires, then stops this Server's metric dispatch workers (if Config.Metrics was set) and waits for them to actually exit too — Shutdown is terminal, a Server cannot accept new connections afterward, so it should honestly release everything it started, not leave metricDispatchWorkers goroutines running forever. Repeated calls are safe.

func (*Server) Snapshot added in v0.2.0 

func (s *Server) Snapshot() ServerSnapshot

Snapshot returns the server's current status. It is safe to call from any goroutine and does not block on session I/O.

type ServerSnapshot added in v0.2.0 

type ServerSnapshot struct {
	ActiveSessions int
	ShuttingDown   bool
	Sessions       []SessionInfo
}

ServerSnapshot is a point-in-time view of server status, suitable for a health or readiness endpoint the application wires up itself.

type Session 

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

Session represents one server-owned Charging Station WebSocket connection. Applications may inspect it, close it, and pass it to Call or typed profile methods. Raw frame writes are intentionally not exposed.

func (*Session) Close 

func (s *Session) Close() error

func (*Session) Context 

func (s *Session) Context() context.Context

func (*Session) Done 

func (s *Session) Done() <-chan struct{}

func (*Session) Err 

func (s *Session) Err() error

func (*Session) Identity 

func (s *Session) Identity() string

func (*Session) Info 

func (s *Session) Info() SessionInfo

func (*Session) Principal 

func (s *Session) Principal() Principal

func (*Session) Version 

func (s *Session) Version() protocol.Version

type SessionInfo 

type SessionInfo struct {
	Identity       string
	Version        protocol.Version
	PrincipalID    string
	RemoteAddr     string
	ConnectedAt    time.Time
	LastReceivedAt time.Time
	LastSentAt     time.Time
	LastPongAt     time.Time
	State          SessionState
}

type SessionState 

type SessionState uint32
const (
	SessionActive SessionState = iota + 1
	SessionClosing
	SessionClosed
)

type TypedAfterHandler added in v0.2.0 

type TypedAfterHandler[Request protocol.Payload, Confirmation protocol.Payload] func(
	context.Context,
	*Session,
	Request,
	Confirmation,
) error

TypedAfterHandler observes a successfully answered CALL after its CALLRESULT has been written to the peer.

type TypedHandler 

type TypedHandler[Request protocol.Payload, Confirmation protocol.Payload] func(
	context.Context,
	*Session,
	Request,
) (Confirmation, error)

TypedHandler receives a schema-validated request and returns a confirmation of the same OCPP version and action.

type TypedSendHandler 

type TypedSendHandler[Request protocol.Payload] func(context.Context, *Session, Request) error

type UniqueIDGenerator 

type UniqueIDGenerator func() string

UniqueIDGenerator returns an OCPP-J unique ID of 1 to 36 characters. Panics, empty IDs, overlong IDs, and duplicate in-flight IDs are converted to errors by Call.

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.