Helper class for sending stream data to a CAN interface. More...

#include <StreamSender.hxx>

Inheritance diagram for openlcb::StreamSenderCan:

Public Member Functions
	StreamSenderCan (Service service, IfCan iface)

StreamSenderCan &	start_stream (Node *src, NodeHandle dst, uint8_t source_stream_id, uint8_t dst_stream_id=StreamDefs::INVALID_STREAM_ID)
	Initiates using the stream sender.

void	close_stream (uint16_t error_code=0)
	Closes the stream when all the bytes are transferred.

StreamSenderCan &	set_proposed_window_size (uint16_t window_size)
	Specifies what the source should propose as window size to the destination.

StreamSenderCan &	set_stream_uid (NodeID stream_uid)
	Specifies the Stream UID to send in the stream initiate request.

void	clear ()
	Sets the stream sender to be available for reuse after a stream has been closed or reached error.

StreamSenderState	get_state ()

uint16_t	get_error ()

uint8_t	get_src_stream_id ()

uint8_t	get_dst_stream_id ()

Action	entry () override
	Start of state machine, called when a buffer of data to send arrives from the application layer.

Public Member Functions inherited from openlcb::StreamSender
	StreamSender (Service *s)

Public Member Functions inherited from StateFlow< ByteBuffer, QList< 1 > >
	StateFlow (Service *service)
	Constructor.

Public Member Functions inherited from TypedStateFlow< MessageType, Base >
	TypedStateFlow (Service *service)
	Constructor.

virtual	~TypedStateFlow ()
	Destructor.

void	send (MessageType *msg, unsigned priority=UINT_MAX) OVERRIDE
	Sends a message to the state flow for processing.

Public Member Functions inherited from FlowInterface< MessageType >
virtual Pool *	pool ()

virtual MessageType *	type_helper ()
	This function is never user in the code, but GDB can use it to infer the correct message types.

MessageType *	alloc ()
	Synchronously allocates a message buffer from the pool of this flow.

void	alloc_async (Executable *target)
	Asynchronously allocates a message buffer from the pool of this flow.

Private Member Functions
void	trigger ()
	Sends an empty message to *this, thereby waking up the state machine.

Action	initiate_stream ()
	Allocates a GenMessage buffer and sends out the stream initiate message to the destination.

Action	send_init_stream ()
	Sends the stream initiate message.

void	stream_initiate_replied (Buffer< GenMessage > *message)
	Callback from GenHandler when a stream initiate reply message arrives at the local interface.

Action	received_init_stream ()
	State executed after wakeup from the stream initiate reply received handler.

Action	do_close_stream ()
	Allocates a GenMessage buffer and sends out the stream close message to the destination.

Action	send_close_stream ()
	Sends the stream close message.

Action	allocate_can_buffer ()
	Allocates a buffer for a CAN frame (for payload send).

Action	got_frame ()
	Got a buffer for an output frame (payload send).

Action	wait_for_stream_proceed ()
	Starts sleeping until a proceed message arrives.

void	stream_proceed_received (Buffer< GenMessage > *message)
	Callback from the handler flow.

Action	stream_proceed_timeout ()

size_t	compute_next_can_length ()

size_t	remaining ()

uint8_t *	payload ()

void	advance (size_t num_bytes)
	Consumes a certain number of bytes from the beginning of the data to send.

Action	return_error (uint32_t code, string message)

Private Attributes
MessageHandler::GenericHandler	streamProceedHandler_
	Handles incoming stream proceed messages.

MessageHandler::GenericHandler	streamInitiateReplyHandler_
	Handles incoming stream initiate reply messages.

IfCan *	ifCan_
	CAN-bus interface.

Node *	node_ {nullptr}
	Which node are we sending the outgoing data from.

NodeHandle	dst_
	Destination node that we are sending to.

size_t	totalByteCount_ {0}
	How many bytes we have transmitted in this stream so far.

StreamSenderState	state_ {IDLE}
	What state the current class is in.

uint8_t	localStreamId_ {StreamDefs::INVALID_STREAM_ID}
	Stream ID at the source node.

uint8_t	dstStreamId_ {StreamDefs::INVALID_STREAM_ID}
	Stream ID at the destination node.

uint8_t	isLoopbackStream_: 1
	Determines whether the stream transmission is happening to localhost.

uint8_t	sleeping_: 1
	True if we are waiting for the timer.

uint8_t	requestClose_: 1
	1 if there is a pending close request.

uint8_t	requestInit_: 1
	1 if there is a pending initialize request.

uint8_t	streamFlags_ {0}
	Flags from the remote node that we got in stream initiate reply.

uint8_t	streamAdditionalFlags_ {0}
	More flags from the remote node that we got in stream initiate reply.

uint16_t	streamWindowSize_ {StreamDefs::MAX_PAYLOAD}
	Total stream window size.

uint16_t	streamWindowRemaining_ {0}
	Remaining stream window size.

uint32_t	errorCode_ {0}
	When the stream process fails, this variable contains an error code.

LimitedPool	canFramePool_ {CAN_FRAME_ALLOC_SIZE, MAX_FRAMES_IN_FLIGHT}
	Source of buffers for outgoing CAN frames.

StateFlowTimer	timer_ {this}
	Helper object for timeouts.

Static Private Attributes
static constexpr size_t	STREAM_PROCEED_TIMEOUT_SEC = 20
	How many seconds for waiting for a stream proceed before we give up with a timeout.

static constexpr size_t	STREAM_INIT_TIMEOUT_SEC = 20
	How many seconds for waiting for a stream init before we give up with a timeout.

static constexpr size_t	MAX_BYTES_PAYLOAD_PER_CAN_FRAME = 7
	How many bytes payload we can copy into a single CAN frame.

static constexpr size_t	MAX_FRAMES_IN_FLIGHT = 4
	How many CAN frames should we allocate at a given time.

static constexpr size_t	CAN_FRAME_ALLOC_SIZE
	How many bytes the allocation of a single CAN frame should be.

Additional Inherited Members
Public Types inherited from openlcb::StreamSender
enum	StreamSenderState : uint8_t { IDLE , STARTED , INITIATING , RUNNING , FULL , CLOSING , STATE_ERROR }
	Describes the different states in the stream sender. More...

Public Types inherited from TypedStateFlow< MessageType, Base >
typedef Base::Action	Action
	Allows using Action without having StateFlowBase:: prefix in front of it.

Public Types inherited from FlowInterface< MessageType >
typedef MessageType	message_type
	Stores the message template type for external reference.

Static Public Member Functions inherited from FlowInterface< MessageType >
static MessageType *	cast_alloc (QMember *entry)
	Down casts and initializes an asynchronous allocation result to the appropriate flow's buffer type.

Protected Member Functions inherited from TypedStateFlow< MessageType, Base >
void	release () OVERRIDE
	Unrefs the current buffer.

void	return_buffer ()
	For state flows that are operated using invoke_subflow_and_wait this is a way to hand back the buffer to the caller.

MessageType *	message ()

MessageType *	transfer_message ()
	Releases ownership of the current message.

Detailed Description

Helper class for sending stream data to a CAN interface.

Todo:: add progress report API.

Definition at line 81 of file StreamSender.hxx.

Constructor & Destructor Documentation

◆ StreamSenderCan()

openlcb::StreamSenderCan::StreamSenderCan	(	Service *	service,
		IfCan *	iface
	)

inline

Definition at line 84 of file StreamSender.hxx.

Member Function Documentation

◆ advance()

void openlcb::StreamSenderCan::advance ( size_t num_bytes )

inlineprivate

Consumes a certain number of bytes from the beginning of the data to send.

Parameters

num_bytes how much data to consume.

Definition at line 517 of file StreamSender.hxx.

◆ allocate_can_buffer()

Action openlcb::StreamSenderCan::allocate_can_buffer ( )

inlineprivate

Allocates a buffer for a CAN frame (for payload send).

Definition at line 389 of file StreamSender.hxx.

◆ clear()

void openlcb::StreamSenderCan::clear ( )

inline

Sets the stream sender to be available for reuse after a stream has been closed or reached error.

Definition at line 167 of file StreamSender.hxx.

◆ close_stream()

void openlcb::StreamSenderCan::close_stream ( uint16_t error_code = 0 )

inline

Closes the stream when all the bytes are transferred.

Parameters

error_code 0 upon success. This code is intended to be transferred in the stream close message, but that is not yet implemented, because the draft protocol does not have provisions for an error code at close.

Definition at line 133 of file StreamSender.hxx.

◆ compute_next_can_length()

size_t openlcb::StreamSenderCan::compute_next_can_length ( )

inlineprivate

Returns: how many bytes of data we can put into the next CAN frame.

Definition at line 486 of file StreamSender.hxx.

◆ do_close_stream()

Action openlcb::StreamSenderCan::do_close_stream ( )

inlineprivate

Allocates a GenMessage buffer and sends out the stream close message to the destination.

Definition at line 366 of file StreamSender.hxx.

◆ entry()

Action openlcb::StreamSenderCan::entry ( )

inlineoverridevirtual

Start of state machine, called when a buffer of data to send arrives from the application layer.

Implements TypedStateFlow< MessageType, Base >.

Definition at line 216 of file StreamSender.hxx.

◆ get_dst_stream_id()

uint8_t openlcb::StreamSenderCan::get_dst_stream_id ( )

inline

Returns: the stream DID (identifier on the receiving node).

Definition at line 209 of file StreamSender.hxx.

◆ get_error()

uint16_t openlcb::StreamSenderCan::get_error ( )

inline

Returns: the error code if we got a rejection from the remote node.

Definition at line 197 of file StreamSender.hxx.

◆ get_src_stream_id()

uint8_t openlcb::StreamSenderCan::get_src_stream_id ( )

inline

Returns: the stream SID (identifier on this node).

Definition at line 203 of file StreamSender.hxx.

◆ get_state()

StreamSenderState openlcb::StreamSenderCan::get_state ( )

inline

Returns: the state of this stream sender.

Definition at line 191 of file StreamSender.hxx.

◆ got_frame()

Action openlcb::StreamSenderCan::got_frame ( )

inlineprivate

Got a buffer for an output frame (payload send).

Definition at line 396 of file StreamSender.hxx.

◆ initiate_stream()

Action openlcb::StreamSenderCan::initiate_stream ( )

inlineprivate

Allocates a GenMessage buffer and sends out the stream initiate message to the destination.

Definition at line 257 of file StreamSender.hxx.

◆ payload()

uint8_t * openlcb::StreamSenderCan::payload ( )

inlineprivate

Returns: pointer to the beginning of the data to send.

Definition at line 509 of file StreamSender.hxx.

◆ received_init_stream()

Action openlcb::StreamSenderCan::received_init_stream ( )

inlineprivate

State executed after wakeup from the stream initiate reply received handler.

Definition at line 330 of file StreamSender.hxx.

◆ remaining()

size_t openlcb::StreamSenderCan::remaining ( )

inlineprivate

Returns: the number of bytes available in the current chunk.

Definition at line 503 of file StreamSender.hxx.

◆ return_error()

Action openlcb::StreamSenderCan::return_error	(	uint32_t	code,
		string	message
	)

inlineprivate

Definition at line 524 of file StreamSender.hxx.

◆ send_close_stream()

Action openlcb::StreamSenderCan::send_close_stream ( )

inlineprivate

Sends the stream close message.

Definition at line 375 of file StreamSender.hxx.

◆ send_init_stream()

Action openlcb::StreamSenderCan::send_init_stream ( )

inlineprivate

Sends the stream initiate message.

Definition at line 266 of file StreamSender.hxx.

◆ set_proposed_window_size()

StreamSenderCan & openlcb::StreamSenderCan::set_proposed_window_size ( uint16_t window_size )

inline

Specifies what the source should propose as window size to the destination.

May be called only after start_stream.

Parameters

window_size in bytes, what should we propose in the stream initiate call

Definition at line 145 of file StreamSender.hxx.

◆ set_stream_uid()

StreamSenderCan & openlcb::StreamSenderCan::set_stream_uid ( NodeID stream_uid )

inline

Specifies the Stream UID to send in the stream initiate request.

May be called only after start_stream. This function must be used if opening an unannounced stream to a destination.

Parameters

stream_uid a valid 6-byte stream identifier.

Todo:: implement opening unannounced streams.

Definition at line 158 of file StreamSender.hxx.

◆ start_stream()

StreamSenderCan & openlcb::StreamSenderCan::start_stream	(	Node *	src,
		NodeHandle	dst,
		uint8_t	source_stream_id,
		uint8_t	dst_stream_id = `StreamDefs::INVALID_STREAM_ID`
	)

inline

Initiates using the stream sender.

May be called only on idle stream senders.

Parameters

src	Source virtual node on the local interface.
dst	Destination node ID to send the stream to.
source_stream_id	8-bit stream ID to use on the this (the source) side.
dst_stream_id	8-bit stream ID to use on the remote side (the destination).

Returns: *this for calling optional settings API commands.

Definition at line 105 of file StreamSender.hxx.

◆ stream_initiate_replied()

void openlcb::StreamSenderCan::stream_initiate_replied ( Buffer< GenMessage > * message )

inlineprivate

Callback from GenHandler when a stream initiate reply message arrives at the local interface.

Definition at line 289 of file StreamSender.hxx.

◆ stream_proceed_received()

void openlcb::StreamSenderCan::stream_proceed_received ( Buffer< GenMessage > * message )

inlineprivate

Callback from the handler flow.

Todo:: add progress callback API

Definition at line 443 of file StreamSender.hxx.

◆ stream_proceed_timeout()

Action openlcb::StreamSenderCan::stream_proceed_timeout ( )

inlineprivate

Todo:: (balazs.racz) somehow merge these two actions: remember that we timed out and close the stream.

Definition at line 470 of file StreamSender.hxx.

◆ trigger()

void openlcb::StreamSenderCan::trigger ( )

inlineprivate

Sends an empty message to *this, thereby waking up the state machine.

Definition at line 249 of file StreamSender.hxx.

◆ wait_for_stream_proceed()

Action openlcb::StreamSenderCan::wait_for_stream_proceed ( )

inlineprivate

Starts sleeping until a proceed message arrives.

Run this state when streamWindowRemaining_ == 0.

Definition at line 429 of file StreamSender.hxx.

Member Data Documentation

◆ CAN_FRAME_ALLOC_SIZE

constexpr size_t openlcb::StreamSenderCan::CAN_FRAME_ALLOC_SIZE

staticconstexprprivate

Initial value:

=

sizeof(CanFrameWriteFlow::message_type)

FlowInterface< Buffer< CanHubData > >::message_type

Buffer< CanHubData > message_type

Stores the message template type for external reference.

Definition StateFlow.hxx:1120

How many bytes the allocation of a single CAN frame should be.

Definition at line 547 of file StreamSender.hxx.

◆ canFramePool_

LimitedPool openlcb::StreamSenderCan::canFramePool_ {CAN_FRAME_ALLOC_SIZE, MAX_FRAMES_IN_FLIGHT}

private

Source of buffers for outgoing CAN frames.

Limtedpool is allocating and releasing to the mainBufferPool, but blocks when we exceed a certain number of allocations until some buffers get freed.

Definition at line 595 of file StreamSender.hxx.

◆ dst_

NodeHandle openlcb::StreamSenderCan::dst_

private

Destination node that we are sending to.

It is important that the alias is filled in here.

Definition at line 564 of file StreamSender.hxx.

◆ dstStreamId_

uint8_t openlcb::StreamSenderCan::dstStreamId_ {StreamDefs::INVALID_STREAM_ID}

private

Stream ID at the destination node.

Todo:: fill in

Definition at line 572 of file StreamSender.hxx.

◆ errorCode_

uint32_t openlcb::StreamSenderCan::errorCode_ {0}

private

When the stream process fails, this variable contains an error code.

Definition at line 591 of file StreamSender.hxx.

◆ ifCan_

IfCan* openlcb::StreamSenderCan::ifCan_

private

CAN-bus interface.

Definition at line 558 of file StreamSender.hxx.

◆ isLoopbackStream_

uint8_t openlcb::StreamSenderCan::isLoopbackStream_

private

Determines whether the stream transmission is happening to localhost.

Almost never true.

Definition at line 575 of file StreamSender.hxx.

◆ localStreamId_

uint8_t openlcb::StreamSenderCan::localStreamId_ {StreamDefs::INVALID_STREAM_ID}

private

Stream ID at the source node.

Todo:: fill in

Definition at line 570 of file StreamSender.hxx.

◆ MAX_BYTES_PAYLOAD_PER_CAN_FRAME

constexpr size_t openlcb::StreamSenderCan::MAX_BYTES_PAYLOAD_PER_CAN_FRAME = 7

staticconstexprprivate

How many bytes payload we can copy into a single CAN frame.

Definition at line 541 of file StreamSender.hxx.

◆ MAX_FRAMES_IN_FLIGHT

constexpr size_t openlcb::StreamSenderCan::MAX_FRAMES_IN_FLIGHT = 4

staticconstexprprivate

How many CAN frames should we allocate at a given time.

Definition at line 544 of file StreamSender.hxx.

◆ node_

Node* openlcb::StreamSenderCan::node_ {nullptr}

private

Which node are we sending the outgoing data from.

This is a local virtual node.

Definition at line 561 of file StreamSender.hxx.

◆ requestClose_

uint8_t openlcb::StreamSenderCan::requestClose_

private

1 if there is a pending close request.

Definition at line 579 of file StreamSender.hxx.

◆ requestInit_

uint8_t openlcb::StreamSenderCan::requestInit_

private

1 if there is a pending initialize request.

Definition at line 581 of file StreamSender.hxx.

◆ sleeping_

uint8_t openlcb::StreamSenderCan::sleeping_

private

True if we are waiting for the timer.

Definition at line 577 of file StreamSender.hxx.

◆ state_

StreamSenderState openlcb::StreamSenderCan::state_ {IDLE}

private

What state the current class is in.

Definition at line 568 of file StreamSender.hxx.

◆ STREAM_INIT_TIMEOUT_SEC

constexpr size_t openlcb::StreamSenderCan::STREAM_INIT_TIMEOUT_SEC = 20

staticconstexprprivate

How many seconds for waiting for a stream init before we give up with a timeout.

Definition at line 538 of file StreamSender.hxx.

◆ STREAM_PROCEED_TIMEOUT_SEC

constexpr size_t openlcb::StreamSenderCan::STREAM_PROCEED_TIMEOUT_SEC = 20

staticconstexprprivate

How many seconds for waiting for a stream proceed before we give up with a timeout.

Definition at line 534 of file StreamSender.hxx.

◆ streamAdditionalFlags_

uint8_t openlcb::StreamSenderCan::streamAdditionalFlags_ {0}

private

More flags from the remote node that we got in stream initiate reply.

Definition at line 585 of file StreamSender.hxx.

◆ streamFlags_

uint8_t openlcb::StreamSenderCan::streamFlags_ {0}

private

Flags from the remote node that we got in stream initiate reply.

Definition at line 583 of file StreamSender.hxx.

◆ streamInitiateReplyHandler_

MessageHandler::GenericHandler openlcb::StreamSenderCan::streamInitiateReplyHandler_

private

Initial value:

{

this, &StreamSenderCan::stream_initiate_replied}

openlcb::StreamSenderCan::stream_initiate_replied

void stream_initiate_replied(Buffer< GenMessage > *message)

Callback from GenHandler when a stream initiate reply message arrives at the local interface.

Definition StreamSender.hxx:289

Handles incoming stream initiate reply messages.

Definition at line 554 of file StreamSender.hxx.

◆ streamProceedHandler_

MessageHandler::GenericHandler openlcb::StreamSenderCan::streamProceedHandler_

private

Initial value:

{

this, &StreamSenderCan::stream_proceed_received}

openlcb::StreamSenderCan::stream_proceed_received

void stream_proceed_received(Buffer< GenMessage > *message)

Callback from the handler flow.

Definition StreamSender.hxx:443

Handles incoming stream proceed messages.

Definition at line 551 of file StreamSender.hxx.

◆ streamWindowRemaining_

uint16_t openlcb::StreamSenderCan::streamWindowRemaining_ {0}

private

Remaining stream window size.

Todo:: fill in

Definition at line 589 of file StreamSender.hxx.

◆ streamWindowSize_

uint16_t openlcb::StreamSenderCan::streamWindowSize_ {StreamDefs::MAX_PAYLOAD}

private

Total stream window size.

Todo:: fill in

Definition at line 587 of file StreamSender.hxx.

◆ timer_

StateFlowTimer openlcb::StreamSenderCan::timer_ {this}

private

Helper object for timeouts.

Definition at line 597 of file StreamSender.hxx.

◆ totalByteCount_

size_t openlcb::StreamSenderCan::totalByteCount_ {0}

private

How many bytes we have transmitted in this stream so far.

Definition at line 566 of file StreamSender.hxx.

The documentation for this class was generated from the following file:

StreamSender.hxx

Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

◆ StreamSenderCan()

Member Function Documentation

◆ advance()

◆ allocate_can_buffer()

◆ clear()

◆ close_stream()

◆ compute_next_can_length()

◆ do_close_stream()

◆ entry()

◆ get_dst_stream_id()

◆ get_error()

◆ get_src_stream_id()

◆ get_state()

◆ got_frame()

◆ initiate_stream()

◆ payload()

◆ received_init_stream()

◆ remaining()

◆ return_error()

◆ send_close_stream()

◆ send_init_stream()

◆ set_proposed_window_size()

◆ set_stream_uid()

◆ start_stream()

◆ stream_initiate_replied()

◆ stream_proceed_received()

◆ stream_proceed_timeout()

◆ trigger()

◆ wait_for_stream_proceed()

Member Data Documentation

◆ CAN_FRAME_ALLOC_SIZE

◆ canFramePool_

◆ dst_

◆ dstStreamId_

◆ errorCode_

◆ ifCan_

◆ isLoopbackStream_

◆ localStreamId_

◆ MAX_BYTES_PAYLOAD_PER_CAN_FRAME

◆ MAX_FRAMES_IN_FLIGHT

◆ node_

◆ requestClose_

◆ requestInit_

◆ sleeping_

◆ state_

◆ STREAM_INIT_TIMEOUT_SEC

◆ STREAM_PROCEED_TIMEOUT_SEC

◆ streamAdditionalFlags_

◆ streamFlags_

◆ streamInitiateReplyHandler_

◆ streamProceedHandler_

◆ streamWindowRemaining_

◆ streamWindowSize_

◆ timer_

◆ totalByteCount_