ocpp/protocol/endpoint
A pure, sans-IO OCPP-J RPC endpoint state machine.
OCPP-J (part 4) makes both peers symmetric RPC endpoints: either side may
initiate CALLs, and each direction allows at most one unanswered CALL at a
time. This module models one side of one connection as a value: feed it
Msg inputs with update, interpret the returned
Effect outputs (socket writes, application
deliveries, violation reports), and drive timeouts from
next_deadline. No processes, sockets, or clocks are
touched here.
Time convention
Time enters exclusively as the now arguments to init and update:
an absolute timestamp in milliseconds from a monotonic clock chosen by
the adapter.
It is an argument rather than a field on every Msg variant so that each
input observes the clock exactly once and messages stay pure, replayable
data. Before any message is processed, the in-flight call is checked
against now — a call whose deadline has passed (now >= deadline) fails
with TimedOut regardless of which input carried the time, so a late
Tick or any other input suffices to fire a timeout, and a stale Tick
(delivered after the state has moved on) is a harmless no-op.
The single-timer pattern
There are no set-timer/cancel-timer effects. Deadlines live in the model;
next_deadline reports the earliest one, and the adapter keeps a single
timer aimed at it, sending Tick when it fires. Because a deadline is a
field of the state that owns it, moving to a new state cancels it
structurally — exactly like gen_statem’s state timeouts.
Correlation and message ids
Outgoing message ids come from a monotonic counter, rendered as strings —
unique per sender per connection. A CALLRESULT is anonymous on the wire
([3, id, payload]), so the machine keeps the in-flight call’s id, the
caller’s ref, and the action; the stored action selects the response
decoder. A bounded ring of the last 32 inbound call ids answered with a
CALLRESULT lets a CALLRESULTERROR (2.1) be matched to the reply it
rejects — it cannot be assumed to refer to the most recent reply, because
the peer may already have a newer CALL in flight.
Terminal state
SocketClosed fails the in-flight call and every queued call with
Disconnected and makes the machine terminal: afterwards a
CallRequested fails immediately with Disconnected (a caller-supplied
ref must always resolve), and every other input is silently ignored —
the connection is gone, so late wire input and replies have nowhere to go
and are not worth reporting.
Types
Why an outgoing CALL failed.
pub type CallFailure {
PeerError(code: String, description: String)
TimedOut
Disconnected
ResponseDecoderMissing(action: String)
ResponseDecodeFailed(errors: List(decode.DecodeError))
}
Constructors
-
PeerError(code: String, description: String)The peer answered with a CALLERROR. The code is the raw wire string so nonstandard codes survive.
-
TimedOutNo answer arrived within
call_timeout_msof the frame being emitted. -
DisconnectedThe transport closed while the call was pending or queued.
-
ResponseDecoderMissing(action: String)A CALLRESULT arrived but the config has no response decoder for the call’s action, so it could not be interpreted.
-
ResponseDecodeFailed(errors: List(decode.DecodeError))A CALLRESULT arrived but its payload could not be decoded.
Version-specific codecs and policy binding the generic machine to one OCPP
version. req/res are typically a version’s dispatch.Request and
dispatch.Response unions; the encode functions mirror the dispatch
modules’ *_to_json signatures (action name plus payload — the action
part of encode_response is ignored, as CALLRESULT is anonymous on the
wire), and the decode functions mirror *_decoder(action).
pub type Config(req, res) {
Config(
call_timeout_ms: Int,
encode_request: fn(req) -> #(String, json.Json),
decode_request: fn(String) -> Result(decode.Decoder(req), Nil),
encode_response: fn(res) -> #(String, json.Json),
decode_response: fn(String) -> Result(
decode.Decoder(res),
Nil,
),
)
}
Constructors
-
Config( call_timeout_ms: Int, encode_request: fn(req) -> #(String, json.Json), decode_request: fn(String) -> Result(decode.Decoder(req), Nil), encode_response: fn(res) -> #(String, json.Json), decode_response: fn(String) -> Result(decode.Decoder(res), Nil), )Arguments
- call_timeout_ms
-
How long (ms) an outgoing CALL may remain unanswered before it fails with
TimedOut, measured from the moment its frame is emitted. - encode_request
-
Encode an outgoing request as its OCPP action name and JSON payload.
- decode_request
-
Select the request payload decoder for an inbound action name.
Error(Nil)means the action is not supported. - encode_response
-
Encode an outgoing response as its OCPP action name and JSON payload.
- decode_response
-
Select the response payload decoder for the action of the CALL being answered.
Error(Nil)means the action is not supported.
Inputs to the machine. ref values are caller-supplied integers, opaque
to the machine, echoed back on CallSucceeded/CallFailed.
pub type Msg(req, res) {
WireIn(text: String)
CallRequested(ref: Int, request: req)
SendRequested(request: req)
ReplyProvided(
id: String,
reply: Result(res, #(rpc.ErrorCode, String)),
)
Tick
SocketClosed
}
Constructors
-
WireIn(text: String)A text frame arrived from the peer.
-
CallRequested(ref: Int, request: req)The application wants to make a CALL. Queued FIFO if one of our calls is already unanswered.
-
SendRequested(request: req)The application wants to emit a SEND (2.1): unconfirmed, exempt from the one-in-flight rule, never queued, no reply expected.
-
ReplyProvided( id: String, reply: Result(res, #(rpc.ErrorCode, String)), )The application answers the inbound CALL with this id:
Okbecomes a CALLRESULT,Errora CALLERROR. -
TickTime passed; sent by the adapter when the
next_deadlinetimer fires. Carries no data — the clock reading is thenowargument toupdate. -
SocketClosedThe transport closed. Fails all pending and queued calls with
Disconnectedand makes the machine terminal.
Outputs emitted by the machine for the adapter/application to interpret.
pub type Output(req, res) {
WireOut(text: String)
CallSucceeded(ref: Int, response: res)
CallFailed(ref: Int, failure: CallFailure)
CallReceived(id: String, request: req)
SendReceived(request: req)
ReplyRejected(id: String, code: String, description: String)
ViolationDetected(violation: Violation)
}
Constructors
-
WireOut(text: String)Write this text frame to the socket.
-
CallSucceeded(ref: Int, response: res)Our CALL identified by
refwas answered with a decodable CALLRESULT. -
CallFailed(ref: Int, failure: CallFailure)Our CALL identified by
reffailed; seeCallFailurefor how. -
CallReceived(id: String, request: req)The peer initiated a CALL. Answer it (at any later point) with
ReplyProvided(id, ...). -
SendReceived(request: req)The peer emitted a SEND. No reply is possible.
-
ReplyRejected(id: String, code: String, description: String)The peer sent a CALLRESULTERROR (2.1) rejecting the CALLRESULT we previously sent for inbound call
id. Codes arrive as raw strings so nonstandard values survive. -
ViolationDetected(violation: Violation)Something protocol-violating (or spec-ignorable but noteworthy) happened; see
Violation. The connection stays usable — policy on whether to drop it belongs to the caller.
Protocol irregularities surfaced to the caller. None of these change the machine’s state beyond the current transition; whether any of them should end the connection is the caller’s policy decision.
pub type Violation {
MalformedFrame(error: json.DecodeError)
IgnoredMessageType(message_type_id: Int)
UnsupportedCallAction(id: String, action: String)
InvalidCallPayload(
id: String,
action: String,
errors: List(decode.DecodeError),
)
UnsupportedSendAction(action: String)
InvalidSendPayload(
action: String,
errors: List(decode.DecodeError),
)
UnexpectedCallResult(id: String)
UnexpectedCallError(id: String)
UnmatchedCallResultError(
id: String,
code: String,
description: String,
)
UnknownReplyId(id: String)
}
Constructors
-
MalformedFrame(error: json.DecodeError)The frame was not parseable OCPP-J JSON.
-
IgnoredMessageType(message_type_id: Int)A frame with an unrecognized message type id arrived. Part 4 §4.1.3: the receiver SHALL ignore it — low severity, connection continues.
-
UnsupportedCallAction(id: String, action: String)An inbound CALL named an action we do not support. A CALLERROR
NotImplementedwas already emitted alongside this violation. -
InvalidCallPayload( id: String, action: String, errors: List(decode.DecodeError), )An inbound CALL payload failed to decode. A CALLERROR
FormatViolationwas already emitted alongside this violation. -
UnsupportedSendAction(action: String)An inbound SEND named an action we do not support. SENDs are unconfirmed, so no reply was (or may be) sent.
-
InvalidSendPayload( action: String, errors: List(decode.DecodeError), )An inbound SEND payload failed to decode. No reply was sent.
-
UnexpectedCallResult(id: String)A CALLRESULT arrived whose id is not our in-flight call — either never ours, already answered, or already timed out (a too-late response).
-
UnexpectedCallError(id: String)A CALLERROR arrived whose id is not our in-flight call.
-
UnmatchedCallResultError( id: String, code: String, description: String, )A CALLRESULTERROR arrived whose id matches no recently sent CALLRESULT of ours.
-
UnknownReplyId(id: String)The application replied to an inbound call id that is not awaiting an answer (unknown, already answered, or arrived after the socket closed). Nothing was sent.
Values
pub fn init(
config: Config(req, res),
now now: Int,
) -> #(Endpoint(req, res), effect.Effect(Output(req, res)))
Create an idle endpoint for a fresh connection at time now. Emits no
effects — the endpoint itself has nothing to say until a role machine (or
the application) makes the first move — and schedules nothing, so now
is currently unused. It is accepted anyway so that every machine in this
layer shares the init(config, now) shape (the station genuinely needs
the clock at init) and adapters can treat them uniformly.
pub fn next_deadline(
endpoint: Endpoint(req, res),
) -> option.Option(Int)
The earliest instant at which this machine needs a Tick to make
progress: the in-flight call’s timeout, if any. Aim a single timer at it;
re-read after every update.
pub fn update(
endpoint: Endpoint(req, res),
msg: Msg(req, res),
now: Int,
) -> #(Endpoint(req, res), effect.Effect(Output(req, res)))
Advance the machine with one input at time now (absolute milliseconds,
monotonic — see the module doc). Expired deadlines are processed before
the message itself.