diff --git a/src/CanKit.Pro.J1939Tp/IJ1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/IJ1939TpChannel.cs
index 3ee2c15..370059d 100644
--- a/src/CanKit.Pro.J1939Tp/IJ1939TpChannel.cs
+++ b/src/CanKit.Pro.J1939Tp/IJ1939TpChannel.cs
@@ -13,7 +13,7 @@ namespace CanKit.Pro.J1939Tp;
///
///
/// Threading model (SRS FR-TP-034 = FR-TP-016/017 applied to J1939-TP): every session's state
-/// (sequence numbers, remaining bytes, block counters, T1..T4/Th deadlines) lives inside a
+/// (sequence numbers, remaining bytes, block counters, T1..T4 deadlines and the BAM packet spacing) lives inside a
/// single mailbox and is only ever read/written on the
/// actor's loop thread. Callers may invoke or
/// concurrently from any thread; the channel serializes them per
@@ -79,10 +79,13 @@ Task SendCmAsync(uint pgn, byte destinationAddress, ReadOnlyMemory payload
IAsyncEnumerable ReceiveAllAsync(CancellationToken cancellationToken = default);
///
- /// Raised on the actor's loop thread every time a full PDU is reassembled. The same datagram
- /// is also enqueued for / . Handlers
- /// must be lightweight and non-throwing; a throwing handler is caught and surfaced via
- /// .
+ /// Raised every time a full PDU is reassembled, after the datagram has been enqueued for
+ /// / , and on the thread pool, not
+ /// on the actor's loop -- so a handler that waits on this channel synchronously gets the
+ /// datagram rather than deadlocking the actor (#58; as ISO-TP delivers). Two datagrams
+ /// reassembled close together may therefore reach their handlers concurrently, and not
+ /// necessarily in the order they completed; keeps that order.
+ /// A throwing handler is caught and surfaced via .
///
event EventHandler? DatagramReceived;
diff --git a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs
index 796ec2a..54c6cb0 100644
--- a/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs
+++ b/src/CanKit.Pro.J1939Tp/J1939TpChannel.cs
@@ -18,7 +18,7 @@ namespace CanKit.Pro.J1939Tp;
/// Actor-driven that composes on top of the CanKit.Pro L2
/// services: for RX demux and TX confirmation,
/// for single-writer per-session state, and
-/// for the J1939-21 §5.10.2.4 timers T1..T4/Th (SRS
+/// for the J1939-21 §5.10.2.4 timers T1..T4 and the BAM packet spacing (SRS
/// FR-TP-032/034).
///
///
@@ -185,6 +185,13 @@ private static void ValidateSendPayload(uint pgn, int payloadLength)
{
if (pgn > J1939Pgn.MaxValue)
throw new ArgumentOutOfRangeException(nameof(pgn), pgn, "PGN must fit in 18 bits.");
+ // A PDU1 PGN's low byte is 0 (SAE J1939-21); the destination is the address argument,
+ // and a value with the byte set is not a PGN -- as J1939Id.ComposePgn refuses it (#55,
+ // #58). Refused here rather than normalised, so the session is keyed on what the peer
+ // will name in its CTS and EndOfMsgAck.
+ if (((pgn >> 8) & 0xFF) < 240 && (pgn & 0xFF) != 0)
+ throw new ArgumentOutOfRangeException(nameof(pgn), pgn,
+ "A PDU1 PGN (PDU Format < 240) has a low byte of 0; the destination address is a separate argument.");
if (payloadLength < J1939TpFrames.MinTpPayloadLength || payloadLength > J1939TpFrames.MaxTpPayloadLength)
throw new ArgumentOutOfRangeException(nameof(payloadLength), payloadLength,
$"J1939-TP payload length must be in [{J1939TpFrames.MinTpPayloadLength}, {J1939TpFrames.MaxTpPayloadLength}] bytes.");
@@ -486,13 +493,20 @@ private void HandleRxTpCm(byte sa, byte da, byte[] payload)
return;
}
+ // An RTS that allows no packet per CTS can never be served: every CTS this
+ // side may send would be a CTS(0) hold. It is not "no limit" -- that is 0xFF
+ // (§5.10.3.1) -- and no session is opened for it; the originator's T3 closes
+ // its side, as for the other malformed RTS above (#58).
+ if (maxCts == 0)
+ return;
+
// Send CTS for the first block, capping at our advertised max-packets-per-CTS
// and at the peer's own RTS cap (0xFF = "no limit" per §5.10.3.1).
// Build the CTS *before* registering the RX session so a validation failure
// cannot leave a timerless orphan that blocks further CM from this source
// (§5.10.3).
byte cap = _options.MaxPacketsPerCts;
- if (maxCts != 0xFF && maxCts > 0 && maxCts < cap) cap = maxCts;
+ if (maxCts != 0xFF && maxCts < cap) cap = maxCts;
byte block = (byte)Math.Min(cap, totalPackets);
var cts = J1939TpFrames.BuildCts(block, 1, dataPgn);
var session = RxSession.NewCm(sa, dataPgn, totalBytes, totalPackets, cap,
@@ -747,7 +761,7 @@ private void StartTx(TxSessionKey key, byte[] pdu, TaskCompletionSource
{
var bam = J1939TpFrames.BuildBam(pdu.Length, totalPackets, key.Pgn);
// Do not schedule TP.DT until the BAM announce is TX-confirmed. Otherwise a rejected
- // BAM can still complete SendBamAsync after Th once DTs finish (Bugbot 3596183535).
+ // BAM can still complete SendBamAsync after the packet spacing once DTs finish (Bugbot 3596183535).
SendTpCm(bam, destinationAddress: J1939TpFrames.GlobalDestinationAddress, session,
onConfirmed: () => OnBamAnnounceConfirmed(key));
}
@@ -756,10 +770,10 @@ private void StartTx(TxSessionKey key, byte[] pdu, TaskCompletionSource
private void OnBamAnnounceConfirmed(TxSessionKey key)
{
if (!_txSessions.TryGetValue(key, out var session) || session.IsCm) return;
- // BAM sender: hold-off Th between BAM and first DT, then Th between subsequent DTs.
+ // BAM sender: the packet spacing between BAM and first DT, then between subsequent DTs.
session.State = TxStage.SendingDt;
session.NextSn = 1;
- _actor.Schedule(_options.Th, () => TrySendNextBamDt(key));
+ _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(key));
}
private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload)
@@ -785,29 +799,44 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload)
return;
}
- // Expected SN for the *next* block. While the last DT of the current block is still
- // awaiting SendConfirmed, NextSn has not yet advanced — but a fast peer (Virtual
- // loopback) may already have received that DT and emitted the next CTS. Accept a CTS
- // that asks for NextSn + BlockRemaining in that race, and apply it once the block drains.
- int expectedSn = session.State == TxStage.SendingDt && session.BlockRemaining > 0
- ? session.NextSn + session.BlockRemaining
- : (session.NextSn == 0 ? 1 : session.NextSn);
- if (nextSn != expectedSn)
+ // What has gone out: the highest packet ever confirmed (HighestSentSn -- an int, so a
+ // 255-packet message's last packet counts, where the byte NextSn wraps to 0), and,
+ // while a block drains, the one outstanding (NextSn, unconfirmed). A CTS for a
+ // packet at or below that frontier asks for it again; one for the packet right
+ // after it is the next block; anything else is a sequence error (table 7, code 7).
+ // After a partial retransmit the cursor is below the frontier, and packets between
+ // the two were sent -- classified against the frontier, not the cursor (Codex on
+ // #152, three times). While a block drains, a fast peer (Virtual loopback) may
+ // already have received the outstanding DT and asked for the block after it:
+ // accepted, and applied once the block drains.
+ bool midBlock = session.State == TxStage.SendingDt && session.BlockRemaining > 0;
+ int frontier = midBlock ? Math.Max(session.HighestSentSn, session.NextSn) : session.HighestSentSn;
+ int expectedSn = midBlock ? session.NextSn + session.BlockRemaining : session.HighestSentSn + 1;
+ bool retransmit = nextSn > 0 && nextSn <= frontier;
+ if (nextSn != expectedSn && !retransmit)
{
- // A CTS for a packet already sent is a retransmit request, which this stack
- // does not serve -- its limit is reached at once (table 7, code 5); one for a
- // packet beyond the next, or for packet 0, which no message has, is a sequence
- // number nothing can recover from (code 7).
- AbortTx(session, nextSn > 0 && nextSn < expectedSn
- ? J1939TpAbortReason.MaximumRetransmitRequestsReached
- : J1939TpAbortReason.BadSequenceNumber,
+ // A CTS for a packet beyond the next, or for packet 0, which no message has,
+ // is a sequence number nothing can recover from (table 7, code 7).
+ AbortTx(session, J1939TpAbortReason.BadSequenceNumber,
$"Peer requested SN {nextSn} but we expected SN {expectedSn}.");
return;
}
+ if (retransmit)
+ {
+ // A CTS for a packet already sent asks for it again (§5.10.2.4): served, from
+ // that packet on, up to MaxRetransmitRequests times per session; the next one
+ // reaches the limit table 7's code 5 names (#58). The whole PDU is in hand, so
+ // nothing is lost by starting over from an earlier packet.
+ if (session.RetransmitRequests >= _options.MaxRetransmitRequests)
+ {
+ AbortTx(session, J1939TpAbortReason.MaximumRetransmitRequestsReached,
+ $"Peer requested SN {nextSn} again after {session.RetransmitRequests} retransmit request(s), the limit.");
+ return;
+ }
+ session.RetransmitRequests++;
+ }
- int packetsBeforeNextBlock = session.State == TxStage.SendingDt
- ? Math.Max(0, session.NextSn + Math.Max(0, session.BlockRemaining) - 1)
- : Math.Max(0, (session.NextSn == 0 ? 1 : session.NextSn) - 1);
+ int packetsBeforeNextBlock = nextSn - 1;
int totalRemaining = session.TotalPackets - packetsBeforeNextBlock;
if (numPackets > totalRemaining)
{
@@ -816,11 +845,15 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload)
return;
}
- // Early CTS while the current block is still draining: stash and apply on block end.
+ // Early CTS while the current block is still draining: stash and apply on block end
+ // -- or, for a retransmit request, as soon as the outstanding DT is confirmed: the
+ // receiver is missing a packet and every later one it gets meanwhile is out of
+ // sequence to it (Codex on #152).
if (session.State == TxStage.SendingDt && session.BlockRemaining > 0)
{
session.PendingCtsNumPackets = numPackets;
session.PendingCtsNextSn = nextSn;
+ session.PendingCtsIsRetransmit = retransmit;
session.HasPendingCts = true;
session.Deadline?.Dispose();
session.Deadline = null;
@@ -831,6 +864,7 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload)
session.NextSn = nextSn;
session.BlockRemaining = numPackets;
session.HasPendingCts = false;
+ session.LastDtQueued = false; // a retransmit re-queues the last packet
session.Deadline?.Dispose();
session.Deadline = null;
TrySendNextCmDt(key);
@@ -842,7 +876,10 @@ private void HandleRxTxSideResponse(byte sa, uint dataPgn, byte[] payload)
// OnCmDtConfirmed (SendingDt → WaitEom). Do NOT treat NextSn >= TotalPackets alone as
// sufficient: CTS for the final SN sets NextSn to TotalPackets before any DT is queued.
bool lastDtOnWire = session.State == TxStage.SendingDt && session.LastDtQueued;
- if (session.State != TxStage.WaitEom && !lastDtOnWire)
+ // After a retransmit that did not reach the last packet, the originator waits for
+ // a CTS; a receiver that has the rest already sends EndOfMsgAck instead, and every
+ // packet having gone out at least once makes that a valid end (Bugbot on #152).
+ if (session.State != TxStage.WaitEom && !lastDtOnWire && !session.AllPacketsSent)
{
AbortTx(session, J1939TpAbortReason.BadSequenceNumber,
$"Peer sent EndOfMsgAck while TX session was in {session.State} (expected WaitEom).");
@@ -902,8 +939,8 @@ private void OnBamDtConfirmed(TxSessionKey key, byte confirmedSn)
}
session.NextSn = (byte)nextSn;
- // Th hold-off between two consecutive BAM DTs (J1939-21 §5.10.3, 50..200 ms).
- _actor.Schedule(_options.Th, () => TrySendNextBamDt(key));
+ // The spacing between two consecutive BAM DTs (J1939-21 §5.10.3, 50..200 ms).
+ _actor.Schedule(_options.BamPacketSpacing, () => TrySendNextBamDt(key));
}
private void TrySendNextCmDt(TxSessionKey key)
@@ -929,10 +966,21 @@ private void OnCmDtConfirmed(TxSessionKey key, byte confirmedSn)
session.NextSn = (byte)nextSn;
session.BlockRemaining--;
int sentPackets = confirmedSn;
+ if (confirmedSn > session.HighestSentSn) session.HighestSentSn = confirmedSn;
+ if (session.HighestSentSn >= session.TotalPackets) session.AllPacketsSent = true;
- if (sentPackets >= session.TotalPackets)
+ // A retransmit request stashed while this block was draining takes effect now, with
+ // the outstanding DT confirmed, rather than after the block (Codex on #152).
+ if (session.HasPendingCts && session.PendingCtsIsRetransmit)
{
- // Last packet -- wait for EndOfMsgAck (T3).
+ ApplyPendingCts(key, session);
+ return;
+ }
+
+ if (sentPackets >= session.TotalPackets && !session.HasPendingCts)
+ {
+ // Last packet -- wait for EndOfMsgAck (T3). A CTS stashed meanwhile is a
+ // retransmit request, applied below like any block's (#58).
session.State = TxStage.WaitEom;
session.Deadline?.Dispose();
session.Deadline = _deadlines.Arm(_options.T3, () => OnTxT3Expired(key));
@@ -947,13 +995,7 @@ private void OnCmDtConfirmed(TxSessionKey key, byte confirmedSn)
// T2 is the receiver's (#31).
if (session.HasPendingCts)
{
- session.HasPendingCts = false;
- session.State = TxStage.SendingDt;
- session.NextSn = session.PendingCtsNextSn;
- session.BlockRemaining = session.PendingCtsNumPackets;
- session.Deadline?.Dispose();
- session.Deadline = null;
- TrySendNextCmDt(key);
+ ApplyPendingCts(key, session);
return;
}
@@ -969,6 +1011,19 @@ private void OnCmDtConfirmed(TxSessionKey key, byte confirmedSn)
TrySendNextCmDt(key);
}
+ private void ApplyPendingCts(TxSessionKey key, TxSession session)
+ {
+ session.HasPendingCts = false;
+ session.PendingCtsIsRetransmit = false;
+ session.State = TxStage.SendingDt;
+ session.NextSn = session.PendingCtsNextSn;
+ session.BlockRemaining = session.PendingCtsNumPackets;
+ session.LastDtQueued = false;
+ session.Deadline?.Dispose();
+ session.Deadline = null;
+ TrySendNextCmDt(key);
+ }
+
private void OnTxT3Expired(TxSessionKey key)
{
if (!_txSessions.TryGetValue(key, out var session) || !session.IsCm) return;
@@ -1068,15 +1123,26 @@ private void SendControlFrame(uint pgn, byte[] payload, byte destinationAddress,
// =========================================================================================
private void EmitPdu(J1939TpDatagram datagram)
{
- try
- {
- DatagramReceived?.Invoke(this, datagram);
- }
- catch (Exception ex)
- {
- RaiseBackgroundException(ex);
- }
+ // Enqueued first, so ReceiveAsync / ReceiveAllAsync see the datagram even if a
+ // DatagramReceived handler blocks; raised off the actor's loop, so a handler that
+ // waits on this channel -- ReceiveAsync, a send -- cannot deadlock the mailbox. As
+ // IsoTpChannel.EmitPdu (#58).
_pduInbox.Writer.TryWrite(RxInboxItem.FromDatagram(datagram));
+
+ var handler = DatagramReceived;
+ if (handler is null)
+ return;
+ _ = Task.Run(() =>
+ {
+ try
+ {
+ handler.Invoke(this, datagram);
+ }
+ catch (Exception ex)
+ {
+ RaiseBackgroundException(ex);
+ }
+ });
}
///
@@ -1218,6 +1284,21 @@ public TxSession(TxSessionKey key, byte[] pdu, int totalPackets, TaskCompletionS
///
public bool HasPendingCts { get; set; }
public byte PendingCtsNumPackets { get; set; }
+ /// The stashed CTS asks for a packet already sent: applied as soon as the outstanding DT is confirmed (Codex on #152).
+ public bool PendingCtsIsRetransmit { get; set; }
+ /// How many CTS for a packet already sent this session has served (#58).
+ public int RetransmitRequests { get; set; }
+ ///
+ /// The highest packet ever confirmed sent -- the frontier a retransmit request is told
+ /// from the next block by, which the cursor NextSn is not after a partial retransmit
+ /// (Codex on #152). An int: the byte NextSn wraps at 255.
+ ///
+ public int HighestSentSn { get; set; }
+ ///
+ /// Every packet has been confirmed sent at least once: from here on an EndOfMsgAck is a
+ /// valid end whatever block a retransmit left the session in (Bugbot on #152).
+ ///
+ public bool AllPacketsSent { get; set; }
///
/// Set when queues the final TP.DT (SN == TotalPackets).
/// Used to accept an early EndOfMsgAck that races ahead of SendConfirmed → WaitEom.
diff --git a/src/CanKit.Pro.J1939Tp/J1939TpException.cs b/src/CanKit.Pro.J1939Tp/J1939TpException.cs
index f704f8b..a392347 100644
--- a/src/CanKit.Pro.J1939Tp/J1939TpException.cs
+++ b/src/CanKit.Pro.J1939Tp/J1939TpException.cs
@@ -23,7 +23,7 @@ protected J1939TpException(CanKitErrorCode errorCode, string message) : base(err
///
/// Raised when a J1939-21 §5.10.5 Connection Abort is issued or received on a TP.CM session,
-/// or when a BAM/TP.CM session gives up because one of T1/T2/T3/T4/Th expired.
+/// or when a BAM/TP.CM session gives up because one of T1/T2/T3/T4 expired.
///
public sealed class J1939TpAbortException : J1939TpException
{
diff --git a/src/CanKit.Pro.J1939Tp/J1939TpFrames.cs b/src/CanKit.Pro.J1939Tp/J1939TpFrames.cs
index a12a675..27289d9 100644
--- a/src/CanKit.Pro.J1939Tp/J1939TpFrames.cs
+++ b/src/CanKit.Pro.J1939Tp/J1939TpFrames.cs
@@ -170,8 +170,14 @@ public static byte[] BuildDt(byte sn, ReadOnlySpan pdu, int offset)
public static uint ReadDataPgn(ReadOnlySpan tpCmPayload)
{
if (tpCmPayload.Length < 8) throw new ArgumentException("TP.CM payload must be 8 bytes.", nameof(tpCmPayload));
- return ((uint)tpCmPayload[5] | ((uint)tpCmPayload[6] << 8) | ((uint)tpCmPayload[7] << 16))
- & J1939Pgn.MaxValue;
+ uint pgn = ((uint)tpCmPayload[5] | ((uint)tpCmPayload[6] << 8) | ((uint)tpCmPayload[7] << 16))
+ & J1939Pgn.MaxValue;
+ // A PDU1 PGN (PDU Format < 240) has a PDU Specific byte of 0 (SAE J1939-21); a stack
+ // that writes the destination address into it instead names the same group, and its
+ // CTS or EndOfMsgAck must still find the session keyed on the PGN. Normalised here, so
+ // every reader of the field agrees (#58).
+ if (((pgn >> 8) & 0xFF) < 240) pgn &= 0x3FF00u;
+ return pgn;
}
private static void WriteDataPgn(byte[] payload, uint dataPgn)
diff --git a/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs b/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs
index e555eb8..aa0dfe2 100644
--- a/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs
+++ b/src/CanKit.Pro.J1939Tp/J1939TpOptions.cs
@@ -15,10 +15,12 @@ namespace CanKit.Pro.J1939Tp;
/// - = 1250 ms — CTS→first TP.DT timeout at the receiver.
/// - = 1250 ms — RTS→CTS, block→next CTS and last DT→EndOfMsgAck timeout at the originator.
/// - = 1050 ms — TP.CM hold timeout at the originator after CTS(0) (§5.10.2.4).
-/// - = 50 ms — hold-off between two consecutive BAM DTs (50..200 ms).
///
/// The standard's Tr (200 ms) is the time a node has to send a response it owes, not
-/// a timer a peer is held to; this stack answers at once and has no option for it (#31).
+/// a timer a peer is held to; this stack answers at once and has no option for it (#31). Its
+/// Th (500 ms) is the holding time between two CTS(0) messages a responder sends; this stack
+/// sends none and has no option for it either (#144). (50 ms)
+/// is the spacing between two BAM packets, §5.10.3's 50..200 ms -- not a timer of §5.10.2.4.
///
public sealed class J1939TpOptions
{
@@ -46,11 +48,22 @@ public sealed class J1939TpOptions
public TimeSpan T4 { get; init; } = TimeSpan.FromMilliseconds(1050);
///
- /// Th — minimum hold-off between two consecutive BAM TP.DT frames on the wire (§5.10.3
+ /// Minimum spacing between two consecutive BAM TP.DT frames on the wire (§5.10.3
/// "50..200 ms"). Default 50 ms to stay at the lower recommended bound while still gating
- /// against a receiver that cannot keep up.
+ /// against a receiver that cannot keep up. Not the standard's Th, which is the holding time
+ /// between CTS(0) messages and which this stack does not use; this option was named Th
+ /// before #144.
///
- public TimeSpan Th { get; init; } = TimeSpan.FromMilliseconds(50);
+ public TimeSpan BamPacketSpacing { get; init; } = TimeSpan.FromMilliseconds(50);
+
+ ///
+ /// How many times per TP.CM session this originator serves a CTS that asks for a packet it
+ /// has already sent -- a retransmit request (§5.10.2.4); the next one is answered with
+ /// Connection Abort reason 5, "maximum retransmit request limit reached" (table 7). The
+ /// standard names the limit and leaves its value to the implementation. Default 2; 0 serves
+ /// none (#58).
+ ///
+ public int MaxRetransmitRequests { get; init; } = 2;
///
/// TX priority for TP.CM / TP.DT frames sent by this channel (0..7, 0 = highest). J1939-21
@@ -85,14 +98,18 @@ public J1939TpOptions With(
TimeSpan? t2 = null,
TimeSpan? t3 = null,
TimeSpan? t4 = null,
- TimeSpan? th = null,
+ TimeSpan? bamPacketSpacing = null,
byte? priority = null,
byte? maxPacketsPerCts = null,
- int? receiveBufferCapacity = null)
+ int? receiveBufferCapacity = null,
+ int? maxRetransmitRequests = null)
{
if (maxPacketsPerCts is 0)
throw new ArgumentOutOfRangeException(nameof(maxPacketsPerCts), maxPacketsPerCts,
"MaxPacketsPerCts must be in [1, 255]; 0 is not a valid CTS grant size.");
+ if (maxRetransmitRequests < 0)
+ throw new ArgumentOutOfRangeException(nameof(maxRetransmitRequests), maxRetransmitRequests,
+ "MaxRetransmitRequests must be >= 0 (0 serves none).");
return new()
{
@@ -100,10 +117,11 @@ public J1939TpOptions With(
T2 = t2 ?? T2,
T3 = t3 ?? T3,
T4 = t4 ?? T4,
- Th = th ?? Th,
+ BamPacketSpacing = bamPacketSpacing ?? BamPacketSpacing,
Priority = priority ?? Priority,
MaxPacketsPerCts = maxPacketsPerCts ?? MaxPacketsPerCts,
ReceiveBufferCapacity = receiveBufferCapacity ?? ReceiveBufferCapacity,
+ MaxRetransmitRequests = maxRetransmitRequests ?? MaxRetransmitRequests,
};
}
@@ -122,5 +140,8 @@ internal void Validate()
if (ReceiveBufferCapacity < 1)
throw new ArgumentOutOfRangeException(nameof(ReceiveBufferCapacity), ReceiveBufferCapacity,
"ReceiveBufferCapacity must be >= 1.");
+ if (MaxRetransmitRequests < 0)
+ throw new ArgumentOutOfRangeException(nameof(MaxRetransmitRequests), MaxRetransmitRequests,
+ "MaxRetransmitRequests must be >= 0 (0 serves none).");
}
}
diff --git a/src/CanKit.Pro.J1939Tp/README.md b/src/CanKit.Pro.J1939Tp/README.md
index 50cf172..f6482a1 100644
--- a/src/CanKit.Pro.J1939Tp/README.md
+++ b/src/CanKit.Pro.J1939Tp/README.md
@@ -13,13 +13,13 @@ can still change until then. See [Versioning](https://github.com/dborgards/CanKi
TP.DT (Data Transfer) frames carry the segmented payload for both flavors, sequence-numbered from 1 (FR-TP-033). Every session runs on its own actor-owned state and its own set of `IDeadline`s (T1, T2, T3, T4 — FR-TP-032), so multiple sessions can execute in parallel over the same physical bus (FR-TP-034/035) without interfering with each other.
-The timers carry J1939-21 §5.10.2.4's meanings: T1 (750 ms) between TP.DTs at the receiver, T2 (1250 ms) from the receiver's CTS to the first TP.DT of the block, T3 (1250 ms) at the originator for the response it is owed — CTS after RTS, the next CTS after a block, EndOfMsgAck after the last packet — and T4 (1050 ms) after a CTS(0) hold. Tr (200 ms) is the time a node has to *send* a response, not a timer a peer is held to, so there is no option for it (#31); the `Th` option is the BAM inter-packet spacing, not the standard's holding time (#144). Connection Abort carries table 7's reason codes and nothing outside the table (`J1939TpAbortReason`), so a peer stack reads the abort as what happened (#33).
+The timers carry J1939-21 §5.10.2.4's meanings: T1 (750 ms) between TP.DTs at the receiver, T2 (1250 ms) from the receiver's CTS to the first TP.DT of the block, T3 (1250 ms) at the originator for the response it is owed — CTS after RTS, the next CTS after a block, EndOfMsgAck after the last packet — and T4 (1050 ms) after a CTS(0) hold. Tr (200 ms) is the time a node has to *send* a response, not a timer a peer is held to, so there is no option for it (#31); Th (500 ms), the holding time between two CTS(0) messages, is not used by this stack, which sends no CTS(0), so there is no option for it either — the BAM inter-packet spacing of §5.10.3 is `BamPacketSpacing`, named `Th` before #144. Connection Abort carries table 7's reason codes and nothing outside the table (`J1939TpAbortReason`), so a peer stack reads the abort as what happened (#33). A CTS that asks for a packet already sent is served, up to `MaxRetransmitRequests` (default 2) times per session, after which reason 5 goes out; a PDU1 PGN in a TP.CM is read with its low byte cleared, so a stack that writes the destination address there still reaches the session; and an RTS allowing no packet per CTS opens no session (#58).
The channel ships on nuget.org alongside the other `CanKit.Pro.*` L2/L3 building blocks. It re-uses:
- `CanKit.Pro.RawCan` — one `ICanBusService` per channel to demultiplex the TP.CM / TP.DT frames back out of the shared bus stream and to confirm outbound frames.
- `CanKit.Pro.Actor` — one `IProtocolActor` mailbox for single-writer session state.
-- `CanKit.Pro.Reliability` — `IDeadlineScheduler` for T1/T2/T3/T4/Th, cancelled/re-armed on the actor's loop.
+- `CanKit.Pro.Reliability` — `IDeadlineScheduler` for T1/T2/T3/T4 and the BAM packet spacing, cancelled/re-armed on the actor's loop.
- `CanKit.Pro.Addressing` — `J1939Id` / `J1939Pgn` for composing the TP.CM (PGN 0xEC00) and TP.DT (PGN 0xEB00) 29-bit IDs.
## Basic usage
diff --git a/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.J1939Tp.approved.txt b/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.J1939Tp.approved.txt
index ae499ed..a596dd8 100644
--- a/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.J1939Tp.approved.txt
+++ b/tests/CanKit.Pro.Tests/ApiApprovals/CanKit.Pro.J1939Tp.approved.txt
@@ -58,15 +58,16 @@ namespace CanKit.Pro.J1939Tp
public sealed class J1939TpOptions
{
public J1939TpOptions() { }
+ public System.TimeSpan BamPacketSpacing { get; init; }
public byte MaxPacketsPerCts { get; init; }
+ public int MaxRetransmitRequests { get; init; }
public byte Priority { get; init; }
public int ReceiveBufferCapacity { get; init; }
public System.TimeSpan T1 { get; init; }
public System.TimeSpan T2 { get; init; }
public System.TimeSpan T3 { get; init; }
public System.TimeSpan T4 { get; init; }
- public System.TimeSpan Th { get; init; }
- public CanKit.Pro.J1939Tp.J1939TpOptions With(System.TimeSpan? t1 = default, System.TimeSpan? t2 = default, System.TimeSpan? t3 = default, System.TimeSpan? t4 = default, System.TimeSpan? th = default, byte? priority = default, byte? maxPacketsPerCts = default, int? receiveBufferCapacity = default) { }
+ public CanKit.Pro.J1939Tp.J1939TpOptions With(System.TimeSpan? t1 = default, System.TimeSpan? t2 = default, System.TimeSpan? t3 = default, System.TimeSpan? t4 = default, System.TimeSpan? bamPacketSpacing = default, byte? priority = default, byte? maxPacketsPerCts = default, int? receiveBufferCapacity = default, int? maxRetransmitRequests = default) { }
}
public sealed class J1939TpSendRejectedException : CanKit.Pro.J1939Tp.J1939TpException
{
diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939/J1939NodeTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939/J1939NodeTests.cs
index 873f15c..c8d03bd 100644
--- a/tests/CanKit.Pro.Tests/TestCases/J1939/J1939NodeTests.cs
+++ b/tests/CanKit.Pro.Tests/TestCases/J1939/J1939NodeTests.cs
@@ -623,11 +623,11 @@ public async Task Send_LargePayload_UsesJ1939TpBamPath()
// Shorten Th so the multi-frame test runs in <1s while still exercising the timer.
var senderOpts = new J1939NodeOptions(Name(1))
{
- TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)),
+ TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)),
};
var receiverOpts = new J1939NodeOptions(Name(2))
{
- TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)),
+ TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)),
};
using var sender = J1939Node.Open(busA, senderOpts);
@@ -680,7 +680,7 @@ public async Task DirectedTpCm_ToClaimedAddress_IsReceivedAfterClaim()
// and surfaces the directed multi-frame PDU on MessageReceived.
using var receiver = J1939Node.Open(busB, new J1939NodeOptions(Name(2))
{
- TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)),
+ TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)),
});
await receiver.ClaimAddressAsync(0xA0).WithTimeout(ShortTimeout);
receiver.ClaimState.Should().Be(J1939ClaimState.Claimed);
@@ -690,7 +690,7 @@ public async Task DirectedTpCm_ToClaimedAddress_IsReceivedAfterClaim()
// J1939-TP channel from a different SA so the frames actually travel across the
// virtual bus and hit the node's transport RX filter.
using var peerTp = CanKit.Pro.J1939Tp.J1939Tp.Open(busA, sourceAddress: 0x55,
- new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)));
+ new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)));
var payload = new byte[24];
for (int i = 0; i < payload.Length; i++) payload[i] = (byte)(0xB0 + i);
@@ -1014,7 +1014,7 @@ public async Task RebindTransport_DoesNotDeliverBamMoreThanOncePerRebind()
var opts = new J1939NodeOptions(Name(1))
{
ClaimAnnounceTimeout = TimeSpan.FromMilliseconds(40),
- TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(2)),
+ TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(2)),
};
using var node = J1939Node.Open(busNode, opts);
@@ -1042,11 +1042,11 @@ static byte[] Datagram(int seq)
}
using var peerTp = CanKit.Pro.J1939Tp.J1939Tp.Open(busPeer, sourceAddress: 0x77,
- new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(2)));
+ new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(2)));
// A second source address for the probes: one SA may only run one BAM session at a
// time, and the probe must not have to queue behind the background stream.
using var probeTp = CanKit.Pro.J1939Tp.J1939Tp.Open(busProbe, sourceAddress: 0x78,
- new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(2)));
+ new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(2)));
int sent = 0;
using var peerCts = new CancellationTokenSource();
@@ -1156,7 +1156,7 @@ public async Task Send_InFlightAcrossReclaim_FailsWithNoAddressException()
var opts = new J1939NodeOptions(Name(1))
{
ClaimAnnounceTimeout = TimeSpan.FromMilliseconds(200),
- TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(60)),
+ TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(60)),
};
using var node = J1939Node.Open(busA, opts);
await node.ClaimAddressAsync(0x11).WithTimeout(ShortTimeout);
@@ -1393,7 +1393,7 @@ public async Task StartPeriodicSend_MultiFrame_Emits_On_An_Exact_Grid_On_A_Clock
var nodeOptions = new J1939NodeOptions(Name(1))
{
- TransportOptions = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(1)),
+ TransportOptions = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(1)),
};
var senderActor = clock.NewActor();
using var sender = new J1939NodeImpl(service, nodeOptions, ownsService: false, senderActor);
diff --git a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs
index 766ea11..8c78d8b 100644
--- a/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs
+++ b/tests/CanKit.Pro.Tests/TestCases/J1939TpTests.cs
@@ -61,7 +61,7 @@ public async Task Two_Channels_Sharing_One_Service_Still_Hear_Each_Other_On_A_Fl
using var bus = ControllableBus.EchoCapable(NewSession());
using var service = new CanBusService(bus);
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(service, sourceAddress: 0x10, options: opts);
using var receiver = J1939TpFactory.Open(service, sourceAddress: 0x20, options: opts);
@@ -98,7 +98,7 @@ public async Task Bam_Sender_On_An_Unflagged_Echo_Bus_Does_Not_Receive_Its_Own_B
var session = NewSession();
using var bus = VirtualAdapterFixture.Open(session, 0, ChannelWorkMode.Echo);
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(bus, sourceAddress: 0x11, options: opts);
var payload = RandomPayload(100, seed: 23);
@@ -125,7 +125,7 @@ public async Task Bam_Roundtrip_ReceiverReassemblesIdenticalPayload()
using var busB = Open(session, 1);
// Shorten Th so the test runs in <1s while still exercising the timer.
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(busA, sourceAddress: 0x11, options: opts);
using var receiver = J1939TpFactory.Open(busB, sourceAddress: 0x22, options: opts);
@@ -187,7 +187,7 @@ public async Task Cm_ExactBoundaryPayload_Reassembles()
var payload = RandomPayload(112, seed: 99);
var receiveTask = receiver.ReceiveAsync().AsTaskWithTimeout(ShortTimeout);
- await sender.SendCmAsync(0xEF10, destinationAddress: 0x04, payload).WithTimeout(ShortTimeout);
+ await sender.SendCmAsync(0xFF10, destinationAddress: 0x04, payload).WithTimeout(ShortTimeout);
var datagram = await receiveTask;
datagram.Payload.Should().Equal(payload);
}
@@ -209,7 +209,7 @@ public async Task Parallel_Bam_And_TwoCm_Sessions_Do_Not_Interfere()
// pacing the test cannot avoid -- inside a 5 s ShortTimeout. That leaves roughly 107 ms
// of slack per scheduled hop, and the gaps are actor Schedule callbacks, so a loaded
// runner eats it. At 5 ms the same 28 gaps cost 140 ms and the margin is ~35x.
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(busA, sourceAddress: 0x10, options: opts);
using var receiverB = J1939TpFactory.Open(busB, sourceAddress: 0xB0, options: opts);
using var receiverC = J1939TpFactory.Open(busC, sourceAddress: 0xC0, options: opts);
@@ -219,8 +219,8 @@ public async Task Parallel_Bam_And_TwoCm_Sessions_Do_Not_Interfere()
var payloadCmC = RandomPayload(250, seed: 3);
var pgnBam = 0xFEF0u;
- var pgnCmB = 0xEE10u;
- var pgnCmC = 0xEE20u;
+ var pgnCmB = 0xFE10u;
+ var pgnCmC = 0xFE20u;
// Collect BAM on both receivers, CM only on its target.
var collectB = CollectAsync(receiverB, count: 2, ShortTimeout);
@@ -269,13 +269,13 @@ public async Task Cm_NoPeer_TimesOutWithAbortException()
using var sender = J1939TpFactory.Open(bus, sourceAddress: 0x30, options: opts);
- var send = sender.SendCmAsync(0xEE30, destinationAddress: 0x99,
+ var send = sender.SendCmAsync(0xFE30, destinationAddress: 0x99,
RandomPayload(50, seed: 5));
Func act = async () => await send.WithTimeout(ShortTimeout);
var ex = (await act.Should().ThrowAsync()).Which;
ex.Reason.Should().Be(J1939TpAbortReason.Timeout);
- ex.Pgn.Should().Be(0xEE30u);
+ ex.Pgn.Should().Be(0xFE30u);
}
// FR-TP-030 lower-bound check: a 9-byte payload (the smallest legal J1939-TP payload;
@@ -287,7 +287,7 @@ public async Task Bam_MinimumPayload_Roundtrip()
using var busA = Open(session, 0);
using var busB = Open(session, 1);
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(busA, sourceAddress: 0x40, options: opts);
using var receiver = J1939TpFactory.Open(busB, sourceAddress: 0x41, options: opts);
@@ -317,7 +317,7 @@ public async Task Bam_MaximumPayload_Roundtrip()
// parking, and the same load costs 1-3 s (#114).
//
// Paced BAM is covered by the other BAM tests in this file, which is why it can go here.
- var opts = new J1939TpOptions().With(th: TimeSpan.Zero);
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.Zero);
using var sender = J1939TpFactory.Open(busA, sourceAddress: 0x50, options: opts);
using var receiver = J1939TpFactory.Open(busB, sourceAddress: 0x51, options: opts);
@@ -344,8 +344,8 @@ public async Task SecondRtsFromSamePeer_DifferentPgn_IsAbortedAndDtRoutesToActiv
const byte receiverSa = 0x22;
const byte peerSa = 0x11;
- const uint activePgn = 0xABCDu;
- const uint intruderPgn = 0x9876u;
+ const uint activePgn = 0xFBCDu;
+ const uint intruderPgn = 0xF876u;
// 14-byte payload = exactly 2 TP.DT frames -> minimal, deterministic size.
var payload = RandomPayload(14, seed: 314);
@@ -469,8 +469,8 @@ public async Task TwoPeers_ConcurrentCm_DifferentPgns_EachReassembledByPeerSa()
const byte receiverSa = 0x22;
const byte peerASa = 0x11;
const byte peerBSa = 0x33;
- const uint pgnA = 0xEE10u;
- const uint pgnB = 0xEE20u;
+ const uint pgnA = 0xFE10u;
+ const uint pgnB = 0xFE20u;
var payloadA = RandomPayload(14, seed: 1);
var payloadB = RandomPayload(14, seed: 2);
@@ -519,7 +519,7 @@ public async Task Bam_AnnounceTxRejected_FailsSendAndDoesNotEmitDt()
using var inner = new CanBusService(busA);
using var rejecting = new RejectTpCmBusService(inner);
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(rejecting, sourceAddress: 0x51, options: opts, leaveOpen: true);
var dtSeen = 0;
@@ -568,7 +568,7 @@ public async Task SendCm_CanceledBeforeStart_DoesNotTransmit()
using var cts = new CancellationTokenSource();
cts.Cancel();
- Func act = async () => await sender.SendCmAsync(0xEE61, destinationAddress: 0x62,
+ Func act = async () => await sender.SendCmAsync(0xFE61, destinationAddress: 0x62,
RandomPayload(50, seed: 61), cts.Token);
await act.Should().ThrowAsync();
@@ -587,7 +587,7 @@ public async Task SendCm_CancelInFlight_SendsConnectionAbort()
const byte senderSa = 0x71;
const byte peerSa = 0x72;
- const uint pgn = 0xEE71u;
+ const uint pgn = 0xFE71u;
using var sender = J1939TpFactory.Open(senderBus, sourceAddress: senderSa);
@@ -655,7 +655,7 @@ public async Task Cm_Receiver_T2Timeout_AbortsWhenNoDtAfterCts()
const byte receiverSa = 0x82;
const byte peerSa = 0x83;
- const uint pgn = 0xEE82u;
+ const uint pgn = 0xFE82u;
var opts = new J1939TpOptions().With(
t2: TimeSpan.FromMilliseconds(80),
@@ -711,7 +711,7 @@ public async Task Cm_Receiver_BadDtSequence_FaultsReceiveAsync()
const byte receiverSa = 0x84;
const byte peerSa = 0x85;
- const uint pgn = 0xEE84u;
+ const uint pgn = 0xFE84u;
var payload = RandomPayload(14, seed: 99);
using var receiver = J1939TpFactory.Open(receiverBus, sourceAddress: receiverSa);
@@ -760,7 +760,7 @@ public async Task Cm_Receiver_BadDtSequence_FaultsReceiveAsync()
bgEx.Reason.Should().Be(J1939TpAbortReason.BadSequenceNumber);
// Channel remains usable for a subsequent BAM after the abort (fault consumed once).
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var senderBus = Open(session, 2);
using var sender = J1939TpFactory.Open(senderBus, sourceAddress: 0x11, options: opts);
var okPayload = RandomPayload(14, seed: 123);
@@ -777,7 +777,7 @@ public async Task Cm_Receiver_BadDtSequence_FaultsReceiveAsync()
[Fact]
public void ReadDataPgn_MasksReservedBitsInByte7()
{
- const uint pgn = 0x12345u; // fits in 18 bits
+ const uint pgn = 0x1F345u; // fits in 18 bits
var rts = J1939TpFrames.BuildRts(totalBytes: 14, totalPackets: 2, maxPacketsPerCts: 0xFF, dataPgn: pgn);
rts[7] |= 0xFC; // set reserved upper 6 bits (would yield > MaxValue if unmasked)
@@ -795,7 +795,7 @@ public async Task Cm_Receiver_RtsWithReservedPgnBits_RepliesCtsAndArmsT2()
const byte receiverSa = 0x88;
const byte peerSa = 0x89;
- const uint pgn = 0xEE88u;
+ const uint pgn = 0xFE88u;
var opts = new J1939TpOptions().With(
t2: TimeSpan.FromMilliseconds(80),
@@ -886,7 +886,7 @@ public async Task Cm_Sender_PeerAbort_FailsSendImmediately()
const byte senderSa = 0x8D;
const byte peerSa = 0x8E;
- const uint pgn = 0xEE8Du;
+ const uint pgn = 0xFE8Du;
var payload = RandomPayload(50, seed: 0x8D);
// Long T3 so a missed abort would hang well past ShortTimeout.
@@ -929,7 +929,7 @@ public async Task Cm_Sender_PrematureEom_FailsSend()
const byte senderSa = 0x91;
const byte peerSa = 0x92;
- const uint pgn = 0xEE91u;
+ const uint pgn = 0xFE91u;
var payload = RandomPayload(14, seed: 91);
using var sender = J1939TpFactory.Open(senderBus, sourceAddress: senderSa);
@@ -976,7 +976,7 @@ public async Task Cm_Sender_EomSizeMismatch_FailsSend()
const byte senderSa = 0x93;
const byte peerSa = 0x94;
- const uint pgn = 0xEE93u;
+ const uint pgn = 0xFE93u;
var payload = RandomPayload(14, seed: 93); // 2 packets
using var sender = J1939TpFactory.Open(senderBus, sourceAddress: senderSa);
@@ -1136,7 +1136,7 @@ public async Task Cm_Receiver_T1Timeout_AbortsWhenDtStopsMidBlock()
const byte receiverSa = 0x92;
const byte peerSa = 0x93;
- const uint pgn = 0xEE92u;
+ const uint pgn = 0xFE92u;
var opts = new J1939TpOptions().With(
t1: TimeSpan.FromMilliseconds(120),
@@ -1192,7 +1192,7 @@ public async Task Cm_Sender_T3Timeout_WhenFollowUpCtsMissing()
const byte senderSa = 0x01;
const byte peerSa = 0x02;
- const uint pgn = 0xEF01u;
+ const uint pgn = 0xFF01u;
// T3 is the originator's timer after the last packet of a block as well as after the
// RTS (§5.10.2.4, #31); T2 is the receiver's and plays no part on this side.
@@ -1241,7 +1241,7 @@ public async Task Cm_Sender_T3Timeout_WhenEomAckMissing()
const byte senderSa = 0x03;
const byte peerSa = 0x04;
- const uint pgn = 0xEF03u;
+ const uint pgn = 0xFF03u;
var opts = new J1939TpOptions().With(
t2: TimeSpan.FromSeconds(5),
@@ -1288,7 +1288,7 @@ public async Task Cm_Sender_T4Timeout_WhenPeerHoldsWithCtsZero()
const byte senderSa = 0x05;
const byte peerSa = 0x06;
- const uint pgn = 0xEF05u;
+ const uint pgn = 0xFF05u;
var opts = new J1939TpOptions().With(
t2: TimeSpan.FromSeconds(5),
@@ -1343,7 +1343,7 @@ public async Task Cm_Receiver_CapsCtsGrant_AtPeerRtsMaximum()
const byte receiverSa = 0x96;
const byte peerSa = 0x97;
- const uint pgn = 0xEE96u;
+ const uint pgn = 0xFE96u;
using var receiver = J1939TpFactory.Open(receiverBus, sourceAddress: receiverSa); // cap 16 by default
@@ -1406,8 +1406,8 @@ public async Task Rts_To_The_Global_Address_Does_Not_Open_A_Session()
const byte receiverSa = 0x22;
const byte peerSa = 0x11;
- const uint globalPgn = 0xABCDu;
- const uint directedPgn = 0x9876u;
+ const uint globalPgn = 0xFBCDu;
+ const uint directedPgn = 0xF876u;
using var receiver = J1939TpFactory.Open(receiverBus, sourceAddress: receiverSa,
options: new J1939TpOptions().With(t2: TimeSpan.FromSeconds(5)));
@@ -1541,7 +1541,7 @@ public async Task Parallel_Bam_Sends_Are_Transmitted_One_After_Another()
var firstPayload = RandomPayload(21, seed: 321); // 3 TP.DT
var secondPayload = RandomPayload(35, seed: 322); // 5 TP.DT
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(senderBus, sourceAddress: senderSa, options: opts);
using var receiver = J1939TpFactory.Open(receiverBus, sourceAddress: 0x20, options: opts);
@@ -1588,7 +1588,7 @@ public async Task A_Queued_Send_Can_Be_Cancelled_Before_It_Reaches_The_Wire()
using var receiverBus = Open(session, 1);
const byte senderSa = 0x10;
- var opts = new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5));
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
using var sender = J1939TpFactory.Open(senderBus, sourceAddress: senderSa, options: opts);
using var receiver = J1939TpFactory.Open(receiverBus, sourceAddress: 0x20, options: opts);
@@ -1625,7 +1625,7 @@ public async Task A_Second_Send_For_The_Same_Destination_And_Pgn_Is_Refused_Whil
var session = NewSession();
using var senderBus = Open(session, 0);
using var sender = J1939TpFactory.Open(senderBus, sourceAddress: 0x10,
- options: new J1939TpOptions().With(th: TimeSpan.FromMilliseconds(5)));
+ options: new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5)));
var first = sender.SendBamAsync(0xFEC1u, RandomPayload(35, seed: 1));
var waiting = sender.SendBamAsync(0xFEC2u, RandomPayload(21, seed: 2));
@@ -1657,9 +1657,11 @@ public void Abort_Reason_Goes_On_The_Wire_As_Table_7_Assigns_It(J1939TpAbortReas
frame[1].Should().Be(code);
}
- // A CTS asking for a packet already sent is a retransmit request; this stack does not
- // retransmit, so the limit is reached at once (table 7, code 5). Before #33 it went out
- // as 5 by coincidence of a different meaning ("unexpected CTS sequence number").
+ // A CTS asking for a packet already sent is a retransmit request; with
+ // MaxRetransmitRequests = 0 this stack serves none, so the limit is reached at once
+ // (table 7, code 5). Before #33 it went out as 5 by coincidence of a different meaning
+ // ("unexpected CTS sequence number"); since #58 the default serves two, covered by
+ // A_Retransmit_Request_Is_Served_Until_The_Limit.
[Fact]
public async Task Cm_Sender_CtsForAPacketAlreadySent_AbortsWithRetransmitLimit()
{
@@ -1672,7 +1674,8 @@ public async Task Cm_Sender_CtsForAPacketAlreadySent_AbortsWithRetransmitLimit()
const uint pgn = 0xFEA1u;
var payload = RandomPayload(21, seed: 161); // 3 packets
- using var sender = J1939TpFactory.Open(senderBus, sourceAddress: senderSa);
+ using var sender = J1939TpFactory.Open(senderBus, sourceAddress: senderSa,
+ options: new J1939TpOptions().With(maxRetransmitRequests: 0));
var rtsSeen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
var firstDtSeen = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
@@ -1891,6 +1894,402 @@ private static async Task SendAndForgetAsync(IJ1939TpChannel send
return new WeakReference(send);
}
+ // ---------------------------------------------------------------------------------------
+ // #58: the four transport findings from the repository review.
+ // ---------------------------------------------------------------------------------------
+
+ // A raw peer on its own bus, observing every TP.CM the subject emits and able to answer.
+ private sealed class RawPeer : IDisposable
+ {
+ private readonly List _cm = new();
+ private readonly List _dt = new();
+ private readonly SemaphoreSlim _ready = new(0);
+ public ICanBus Bus { get; }
+ public byte SubjectSa { get; }
+
+ public RawPeer(ICanBus bus, byte subjectSa)
+ {
+ Bus = bus;
+ SubjectSa = subjectSa;
+ bus.FrameObserved += (_, e) =>
+ {
+ var frame = e.CanFrame;
+ if (!frame.IsExtendedFrame) return;
+ var fields = J1939Id.Decompose((uint)frame.ID);
+ if (fields.SourceAddress != subjectSa) return;
+ var data = frame.Data.ToArray();
+ lock (_cm)
+ {
+ if (J1939Pgn.IsTransportCm(fields.Pgn)) _cm.Add(data);
+ else if (fields.Pgn == J1939Pgn.TpDt) _dt.Add(data);
+ }
+ _ready.Release();
+ };
+ }
+
+ public int DtCount { get { lock (_cm) return _dt.Count; } }
+
+ public async Task WaitForCmAsync(Func predicate, TimeSpan timeout)
+ {
+ using var cts = new CancellationTokenSource(timeout);
+ while (true)
+ {
+ lock (_cm)
+ {
+ foreach (var d in _cm) if (predicate(d)) return d;
+ }
+ await _ready.WaitAsync(cts.Token).ConfigureAwait(false);
+ }
+ }
+
+ public async Task WaitForDtCountAsync(int count, TimeSpan timeout)
+ {
+ using var cts = new CancellationTokenSource(timeout);
+ while (DtCount < count) await _ready.WaitAsync(cts.Token).ConfigureAwait(false);
+ }
+
+ public void SendCm(byte peerSa, byte[] payload)
+ => Bus.Transmit(CanFrame.Classic((int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, SubjectSa), payload, isExtendedFrame: true));
+
+ public void SendDt(byte peerSa, byte[] payload)
+ => Bus.Transmit(CanFrame.Classic((int)J1939Id.ComposePgn(7, J1939Pgn.TpDt, peerSa, SubjectSa), payload, isExtendedFrame: true));
+
+ public void Dispose() => _ready.Dispose();
+ }
+
+ // #58: a peer that writes the destination address into the low byte of a PDU1 PGN in its
+ // TP.CM names the same group; its CTS must still reach the originator's session.
+ [Fact]
+ public async Task A_Cts_With_The_Destination_In_The_Pgns_Low_Byte_Reaches_The_Session()
+ {
+ var session = NewSession();
+ using var subjectBus = Open(session, 0);
+ using var peerBus = Open(session, 1);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xC800u; // PDU1: PF 0xC8, PS 0
+ var payload = RandomPayload(14, seed: 58);
+ using var sender = J1939TpFactory.Open(subjectBus, sourceAddress: subjectSa);
+ using var peer = new RawPeer(peerBus, subjectSa);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlRts, ShortTimeout);
+
+ // The CTS's PGN field carries our address in the low byte: 0xF810 rather than 0xC800.
+ var ctsFrame = J1939TpFrames.BuildCts(numPackets: 2, nextPacketSn: 1, dataPgn: pgn | subjectSa);
+ peer.SendCm(peerSa, ctsFrame);
+ await peer.WaitForDtCountAsync(2, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildEomAck(14, 2, pgn | subjectSa));
+ await send.WaitAsync(ShortTimeout);
+ }
+
+ // #58: a PDU1 PGN with its low byte set is not a PGN -- the destination is the address
+ // argument -- and is refused before anything goes out, as J1939Id.ComposePgn refuses it
+ // (#55); the session is keyed on what the peer names in its CTS.
+ [Fact]
+ public async Task A_Pdu1_Pgn_With_A_Low_Byte_Is_Refused_Before_Anything_Goes_Out()
+ {
+ var session = NewSession();
+ using var bus = Open(session, 0);
+ using var sender = J1939TpFactory.Open(bus, sourceAddress: 0x10);
+ int transmitted = 0;
+ bus.FrameObserved += (_, e) => { if (e.CanFrame.IsExtendedFrame) Interlocked.Increment(ref transmitted); };
+
+ Func cm = () => sender.SendCmAsync(0xEE8Du, 0x20, RandomPayload(14, seed: 1));
+ await cm.Should().ThrowAsync().WithParameterName("pgn");
+ Func bam = () => sender.SendBamAsync(0xEE8Du, RandomPayload(14, seed: 1));
+ await bam.Should().ThrowAsync().WithParameterName("pgn");
+ await Task.Delay(50);
+ transmitted.Should().Be(0, "nothing was transmitted");
+ }
+
+ // #58: an RTS that allows no packet per CTS can never be served -- every CTS would be a
+ // hold -- and is not "no limit"; no session is opened, and the next well-formed RTS from
+ // the same peer is served.
+ [Fact]
+ public async Task An_Rts_Allowing_No_Packet_Per_Cts_Opens_No_Session()
+ {
+ var session = NewSession();
+ using var subjectBus = Open(session, 0);
+ using var peerBus = Open(session, 1);
+ const byte subjectSa = 0x22, peerSa = 0x11;
+ const uint pgn = 0xFBCDu;
+ using var receiver = J1939TpFactory.Open(subjectBus, sourceAddress: subjectSa);
+ using var peer = new RawPeer(peerBus, subjectSa);
+
+ peer.SendCm(peerSa, J1939TpFrames.BuildRts(totalBytes: 14, totalPackets: 2, maxPacketsPerCts: 0, dataPgn: pgn));
+ Func none = () => peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlCts, TimeSpan.FromMilliseconds(300));
+ await none.Should().ThrowAsync("no CTS answers an RTS that permits none");
+
+ // Served, because no session lingers for the malformed one.
+ peer.SendCm(peerSa, J1939TpFrames.BuildRts(totalBytes: 14, totalPackets: 2, maxPacketsPerCts: 0xFF, dataPgn: pgn));
+ var ctsFrame = await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlCts, ShortTimeout);
+ ctsFrame[1].Should().BeGreaterThan(0);
+ }
+
+ // #58: a CTS for a packet already sent asks for it again and is served, up to
+ // MaxRetransmitRequests times; the next one reaches table 7's limit (reason 5).
+ [Fact]
+ public async Task A_Retransmit_Request_Is_Served_Until_The_Limit()
+ {
+ var session = NewSession();
+ using var subjectBus = Open(session, 0);
+ using var peerBus = Open(session, 1);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xFECAu;
+ var payload = RandomPayload(21, seed: 5); // three packets
+ var opts = new J1939TpOptions().With(maxRetransmitRequests: 1);
+ using var sender = J1939TpFactory.Open(subjectBus, sourceAddress: subjectSa, options: opts);
+ using var peer = new RawPeer(peerBus, subjectSa);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlRts, ShortTimeout);
+
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn));
+ await peer.WaitForDtCountAsync(3, ShortTimeout);
+
+ // Packet 2 again: served -- one more DT, with SN 2.
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 2, dataPgn: pgn));
+ await peer.WaitForDtCountAsync(4, ShortTimeout);
+
+ // And again: the limit of one is reached, reason 5.
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 2, dataPgn: pgn));
+ var abort = await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlAbort, ShortTimeout);
+ abort[1].Should().Be((byte)J1939TpAbortReason.MaximumRetransmitRequestsReached);
+ Func failed = () => send;
+ await failed.Should().ThrowAsync();
+ }
+
+ // Bugbot on #152: a receiver that asked for one packet again and then has the whole
+ // message sends EndOfMsgAck, not another CTS; the originator, every packet sent at least
+ // once, completes on it.
+ [Fact]
+ public async Task An_End_Of_Message_After_A_Partial_Retransmit_Completes_The_Send()
+ {
+ var session = NewSession();
+ using var subjectBus = Open(session, 0);
+ using var peerBus = Open(session, 1);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xFEC5u;
+ var payload = RandomPayload(21, seed: 6); // three packets
+ using var sender = J1939TpFactory.Open(subjectBus, sourceAddress: subjectSa);
+ using var peer = new RawPeer(peerBus, subjectSa);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlRts, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn));
+ await peer.WaitForDtCountAsync(3, ShortTimeout);
+
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 2, dataPgn: pgn)); // packet 2 again
+ await peer.WaitForDtCountAsync(4, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildEomAck(21, 3, pgn));
+ await send.WaitAsync(ShortTimeout);
+ }
+
+ // Codex on #152: a retransmit request that arrives while a block is still draining takes
+ // effect as soon as the outstanding DT is confirmed -- the receiver is missing a packet,
+ // and every later one it gets meanwhile is out of sequence to it -- not after the block.
+ [Fact]
+ public async Task A_Retransmit_Request_Mid_Block_Takes_Effect_After_The_Outstanding_Packet()
+ {
+ using var bus = ControllableBus.DeferredEchoCapable(NewSession());
+ using var service = new CanBusService(bus);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xFEC6u;
+ var payload = RandomPayload(21, seed: 7); // three packets
+ using var sender = J1939TpFactory.Open(service, sourceAddress: subjectSa);
+
+ var dtSns = new List();
+ bus.OnTransmitting = f =>
+ {
+ var fields = J1939Id.Decompose((uint)f.ID);
+ if (fields.Pgn == J1939Pgn.TpDt) lock (dtSns) dtSns.Add(f.Data.Span[0]);
+ };
+ static CanFrame PeerCm(byte peerSa, byte subjectSa, byte[] data)
+ => CanFrame.Classic((int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, subjectSa), data, isExtendedFrame: true);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); // the RTS
+ bus.DeferredEchoes.ReleaseNext();
+ bus.RaiseObserved(PeerCm(peerSa, subjectSa, J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn)), isEcho: false);
+ await bus.DeferredEchoes.WaitForEnqueuedAsync(2, ShortTimeout); // DT 1, its confirmation held
+
+ // Packet 1 asked for again while DT 1 is still outstanding.
+ bus.RaiseObserved(PeerCm(peerSa, subjectSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 1, dataPgn: pgn)), isEcho: false);
+ await Task.Delay(50); // the CTS is on the actor before the confirmation is released
+ bus.DeferredEchoes.ReleaseNext();
+ await bus.DeferredEchoes.WaitForEnqueuedAsync(3, ShortTimeout); // the next DT
+
+ byte[] sns;
+ lock (dtSns) sns = dtSns.ToArray();
+ sns.Should().Equal(new byte[] { 1, 1 }, "the retransmit took effect after the outstanding packet, not after the block");
+
+ // The send is left to the channel's disposal; its outcome is not this test's subject.
+ cts.Cancel();
+ Func cancelled = () => send;
+ await cancelled.Should().ThrowAsync();
+ }
+
+ // Codex on #152: while a block drains, "already sent" reaches only up to the outstanding
+ // packet; a CTS for a later packet of the grant would skip the ones between, and is a
+ // sequence error (table 7, code 7), not a retransmit.
+ [Fact]
+ public async Task A_Cts_For_An_Unsent_Packet_Of_The_Block_Is_A_Sequence_Error_Not_A_Retransmit()
+ {
+ using var bus = ControllableBus.DeferredEchoCapable(NewSession());
+ using var service = new CanBusService(bus);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xFEC7u;
+ var payload = RandomPayload(21, seed: 8); // three packets
+ using var sender = J1939TpFactory.Open(service, sourceAddress: subjectSa);
+
+ var aborts = new List();
+ bus.OnTransmitting = f =>
+ {
+ var fields = J1939Id.Decompose((uint)f.ID);
+ if (J1939Pgn.IsTransportCm(fields.Pgn) && f.Data.Span[0] == J1939TpFrames.ControlAbort)
+ lock (aborts) aborts.Add(f.Data.ToArray());
+ };
+ static CanFrame PeerCm(byte peerSa, byte subjectSa, byte[] data)
+ => CanFrame.Classic((int)J1939Id.ComposePgn(7, J1939Pgn.TpCm, peerSa, subjectSa), data, isExtendedFrame: true);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await bus.DeferredEchoes.WaitForEnqueuedAsync(1, ShortTimeout); // the RTS
+ bus.DeferredEchoes.ReleaseNext();
+ bus.RaiseObserved(PeerCm(peerSa, subjectSa, J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn)), isEcho: false);
+ await bus.DeferredEchoes.WaitForEnqueuedAsync(2, ShortTimeout); // DT 1, its confirmation held
+
+ // Packet 3 asked for while DT 1 is outstanding and DT 2 unsent.
+ bus.RaiseObserved(PeerCm(peerSa, subjectSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 3, dataPgn: pgn)), isEcho: false);
+ Func failed = () => send;
+ var ex = await failed.Should().ThrowAsync();
+ ex.Which.Reason.Should().Be(J1939TpAbortReason.BadSequenceNumber);
+ bus.DeferredEchoes.ReleaseNext();
+ await Task.Delay(50);
+ lock (aborts) aborts.Should().ContainSingle().Which[1].Should().Be((byte)J1939TpAbortReason.BadSequenceNumber);
+ }
+
+ // Codex on #152: after a partial retransmit the cursor is below the highest packet sent,
+ // and a request for a packet between the two is a retransmit -- served, and counted.
+ [Fact]
+ public async Task A_Retransmit_Request_Above_The_Cursor_But_Below_The_Highest_Sent_Is_Served()
+ {
+ var session = NewSession();
+ using var subjectBus = Open(session, 0);
+ using var peerBus = Open(session, 1);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xFEC8u;
+ var payload = RandomPayload(21, seed: 9); // three packets
+ using var sender = J1939TpFactory.Open(subjectBus, sourceAddress: subjectSa);
+ using var peer = new RawPeer(peerBus, subjectSa);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlRts, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn));
+ await peer.WaitForDtCountAsync(3, ShortTimeout);
+
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 1, dataPgn: pgn)); // packet 1 again: cursor 2
+ await peer.WaitForDtCountAsync(4, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 3, dataPgn: pgn)); // packet 3 again: above the cursor, sent before
+ await peer.WaitForDtCountAsync(5, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildEomAck(21, 3, pgn));
+ await send.WaitAsync(ShortTimeout);
+ }
+
+ // Codex on #152: and a request for the packet at the cursor after a partial retransmit is
+ // a retransmit too -- not the next block -- and counts against the limit.
+ [Fact]
+ public async Task A_Retransmit_Request_At_The_Cursor_After_A_Partial_Retransmit_Counts()
+ {
+ var session = NewSession();
+ using var subjectBus = Open(session, 0);
+ using var peerBus = Open(session, 1);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xFEC9u;
+ var payload = RandomPayload(21, seed: 10); // three packets
+ var opts = new J1939TpOptions().With(maxRetransmitRequests: 1);
+ using var sender = J1939TpFactory.Open(subjectBus, sourceAddress: subjectSa, options: opts);
+ using var peer = new RawPeer(peerBus, subjectSa);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlRts, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 3, nextPacketSn: 1, dataPgn: pgn));
+ await peer.WaitForDtCountAsync(3, ShortTimeout);
+
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 1, dataPgn: pgn)); // the one retransmit allowed
+ await peer.WaitForDtCountAsync(4, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 2, dataPgn: pgn)); // at the cursor, sent before: the second
+ var abort = await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlAbort, ShortTimeout);
+ abort[1].Should().Be((byte)J1939TpAbortReason.MaximumRetransmitRequestsReached);
+ Func failed = () => send;
+ await failed.Should().ThrowAsync();
+ }
+
+ // Codex on #152: a 255-packet message wraps the byte NextSn to 0 once every packet is
+ // sent; a retransmit request for its last packet must still read as one, and be served.
+ [Fact]
+ public async Task A_Retransmit_Request_For_The_Last_Of_255_Packets_Is_Served()
+ {
+ var session = NewSession();
+ using var subjectBus = Open(session, 0);
+ using var peerBus = Open(session, 1);
+ const byte subjectSa = 0x10, peerSa = 0x20;
+ const uint pgn = 0xFEC0u;
+ var payload = RandomPayload(J1939TpFrames.MaxTpPayloadLength, seed: 255); // 255 packets
+ using var sender = J1939TpFactory.Open(subjectBus, sourceAddress: subjectSa);
+ using var peer = new RawPeer(peerBus, subjectSa);
+
+ using var cts = new CancellationTokenSource(ShortTimeout);
+ var send = sender.SendCmAsync(pgn, peerSa, payload, cts.Token);
+ await peer.WaitForCmAsync(d => d[0] == J1939TpFrames.ControlRts, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 255, nextPacketSn: 1, dataPgn: pgn));
+ await peer.WaitForDtCountAsync(255, ShortTimeout);
+
+ peer.SendCm(peerSa, J1939TpFrames.BuildCts(numPackets: 1, nextPacketSn: 255, dataPgn: pgn));
+ await peer.WaitForDtCountAsync(256, ShortTimeout);
+ peer.SendCm(peerSa, J1939TpFrames.BuildEomAck(J1939TpFrames.MaxTpPayloadLength, 255, pgn));
+ await send.WaitAsync(ShortTimeout);
+ }
+
+ // #58: the datagram is in the inbox before DatagramReceived is raised, and the event is
+ // raised off the actor -- so a handler that waits on ReceiveAsync gets the datagram
+ // rather than deadlocking the channel, as an ISO-TP handler does.
+ [Fact]
+ public async Task DatagramReceived_Finds_The_Datagram_Already_Receivable()
+ {
+ var session = NewSession();
+ using var senderBus = Open(session, 0);
+ using var receiverBus = Open(session, 1);
+ var opts = new J1939TpOptions().With(bamPacketSpacing: TimeSpan.FromMilliseconds(5));
+ using var sender = J1939TpFactory.Open(senderBus, sourceAddress: 0x30, options: opts);
+ using var receiver = J1939TpFactory.Open(receiverBus, sourceAddress: 0x31, options: opts);
+ var payload = RandomPayload(14, seed: 9);
+
+ var fromHandler = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ receiver.DatagramReceived += (_, d) =>
+ {
+ try
+ {
+ // A synchronous wait on the channel from inside its own event.
+ using var wait = new CancellationTokenSource(TimeSpan.FromSeconds(2));
+ fromHandler.TrySetResult(receiver.ReceiveAsync(wait.Token).GetAwaiter().GetResult());
+ }
+ catch (OperationCanceledException ex)
+ {
+ fromHandler.TrySetException(ex); // the wait timed out: the datagram was not receivable
+ }
+ };
+
+ await sender.SendBamAsync(0xFECAu, payload).WaitAsync(ShortTimeout);
+ var datagram = await fromHandler.Task.WaitAsync(ShortTimeout);
+ datagram.Payload.Should().Equal(payload);
+ }
}
///