TripleG3.P2P 1.2.21

TripleG3.P2P

Quick Glance Examples

UDP Example

using System;
using System.Linq;
using System.Net;
using TripleG3.P2P.Attributes;
using TripleG3.P2P.Core;

var peers = new (string Name, IPAddress Address, int Port)[]
{
    ("Client A", IPAddress.Parse("10.42.0.10"), 7000),
    ("Client B", IPAddress.Parse("10.42.0.21"), 7000),
    ("Client C", IPAddress.Parse("10.42.0.23"), 7100)
};

var clientName = args.Length == 1
    ? $"Client {args[0].Trim().ToUpperInvariant()}"
    : throw new ArgumentException("Run with A, B, or C.");

var current = peers.Single(peer => peer.Name == clientName);
var bus = SerialBusFactory.CreateUdp();

bus.SubscribeTo<Chat>(message => Console.WriteLine($"{message.Sender}: {message.Text}"));

await bus.StartListeningAsync(new ProtocolConfiguration
{
    LocalAddress = current.Address,
    LocalPort = current.Port,
    OutboundEndPoints = peers
        .Where(peer => peer.Name != current.Name)
        .Select(peer => new IPEndPoint(peer.Address, peer.Port))
        .ToArray(),
    SerializationProtocol = SerializationProtocol.LengthPrefixed
});

while (Console.ReadLine() is { } text)
{
    await bus.SendAsync(new Chat(current.Name, text));
}

await bus.CloseConnectionAsync();

[P2PMessage("Chat")]
public sealed record Chat(
    [property: P2PProperty(1)] string Sender,
    [property: P2PProperty(2)] string Text);

TCP Example

using System;
using System.Linq;
using System.Net;
using TripleG3.P2P.Attributes;
using TripleG3.P2P.Core;

var peers = new (string Name, IPAddress Address, int Port)[]
{
    ("Client A", IPAddress.Parse("10.42.0.10"), 7000),
    ("Client B", IPAddress.Parse("10.42.0.21"), 7000),
    ("Client C", IPAddress.Parse("10.42.0.23"), 7100)
};

var clientName = args.Length == 1
    ? $"Client {args[0].Trim().ToUpperInvariant()}"
    : throw new ArgumentException("Run with A, B, or C.");

var current = peers.Single(peer => peer.Name == clientName);
var bus = SerialBusFactory.CreateTcp();

bus.SubscribeTo<Chat>(message => Console.WriteLine($"{message.Sender}: {message.Text}"));

await bus.StartListeningAsync(new ProtocolConfiguration
{
    LocalAddress = current.Address,
    LocalPort = current.Port,
    OutboundEndPoints = peers
        .Where(peer => peer.Name != current.Name)
        .Select(peer => new IPEndPoint(peer.Address, peer.Port))
        .ToArray(),
    SerializationProtocol = SerializationProtocol.LengthPrefixed
});

while (Console.ReadLine() is { } text)
{
    await bus.SendAsync(new Chat(current.Name, text));
}

await bus.CloseConnectionAsync();

[P2PMessage("Chat")]
public sealed record Chat(
    [property: P2PProperty(1)] string Sender,
    [property: P2PProperty(2)] string Text);

File Transfer Example

using System;
using System.IO;
using System.Linq;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using TripleG3.P2P.FileTransfer;

if (args.Length is < 1 or > 2)
{
    throw new ArgumentException("Run with A, B, or C and an optional source file path.");
}

var peers = new (string Name, IPEndPoint EndPoint)[]
{
    ("Client A", new IPEndPoint(IPAddress.Parse("10.42.0.10"), 9100)),
    ("Client B", new IPEndPoint(IPAddress.Parse("10.42.0.21"), 9100)),
    ("Client C", new IPEndPoint(IPAddress.Parse("10.42.0.23"), 9110))
};

var clientName = $"Client {args[0].Trim().ToUpperInvariant()}";
var current = peers.Single(peer => peer.Name == clientName);

await using var client = new PeerFileTransferClient(new FileTransferOptions
{
    LocalEndPoint = current.EndPoint
});

client.TransferRequested += (request, _) =>
    new ValueTask<FileTransferDecision>(
        FileTransferDecision.Accept(Path.Combine(Path.GetTempPath(), request.FileName)));

await client.StartAsync();

if (args.Length == 2)
{
    _ = await client.SendAsync(
        args[1],
        peers.Where(peer => peer.Name != current.Name).Select(peer => peer.EndPoint).ToArray());
}

await Task.Delay(Timeout.InfiniteTimeSpan);

RTP Audio Example

using System;
using System.Net;
using TripleG3.P2P.Audio;

var peers = new (string Name, IPEndPoint EndPoint)[]
{
    ("Client A", new IPEndPoint(IPAddress.Parse("10.42.0.10"), 5004)),
    ("Client B", new IPEndPoint(IPAddress.Parse("10.42.0.21"), 5004)),
    ("Client C", new IPEndPoint(IPAddress.Parse("10.42.0.23"), 5014))
};

var clientName = args.Length == 1
    ? $"Client {args[0].Trim().ToUpperInvariant()}"
    : throw new ArgumentException("Run with A, B, or C.");

var currentIndex = Array.FindIndex(peers, peer => peer.Name == clientName);
if (currentIndex < 0)
{
    throw new ArgumentException("Run with A, B, or C.");
}

var current = peers[currentIndex];
var remote = peers[(currentIndex + 1) % peers.Length];
var config = new RtpAudioConfig
{
    LocalAddress = current.EndPoint.Address,
    LocalPort = current.EndPoint.Port,
    RemoteEndPoint = remote.EndPoint,
    Ssrc = 0xA11CEu
};

await using var receiver = new RtpAudioReceiver(config);
await using var sender = new RtpAudioSender(config);

receiver.AudioFrameReceived += frame => Console.WriteLine($"{frame.Timestamp}:{frame.OpusFrame.Length}");

await receiver.StartAsync();

var timestamp = 0u;
while (Console.ReadLine() is not null)
{
    await sender.SendAsync(new byte[] { 0xF8, 0xFF, 0xFE }, new AudioFrameMetadata(timestamp));
    timestamp += 960;
}

await receiver.StopAsync();

RTP Video Example

using System;
using System.IO;
using System.Net;
using System.Threading;
using System.Threading.Tasks;
using TripleG3.P2P.Video;
using TripleG3.P2P.Video.Primitives;

if (args.Length is < 1 or > 2)
{
    throw new ArgumentException("Run with A, B, or C and an optional Annex B H.264 path.");
}

var peers = new (string Name, IPEndPoint EndPoint)[]
{
    ("Client A", new IPEndPoint(IPAddress.Parse("10.42.0.10"), 5006)),
    ("Client B", new IPEndPoint(IPAddress.Parse("10.42.0.21"), 5006)),
    ("Client C", new IPEndPoint(IPAddress.Parse("10.42.0.23"), 5016))
};

var clientName = $"Client {args[0].Trim().ToUpperInvariant()}";
var currentIndex = Array.FindIndex(peers, peer => peer.Name == clientName);
if (currentIndex < 0)
{
    throw new ArgumentException("Run with A, B, or C and an optional Annex B H.264 path.");
}

var current = peers[currentIndex];
var remote = peers[(currentIndex + 1) % peers.Length];
const uint ssrc = 0xA11CEu;

await using var receiver = new RtpVideoReceiver(
    new RtpVideoReceiverConfig
    {
        LocalAddress = current.EndPoint.Address,
        LocalPort = current.EndPoint.Port,
        ExpectedSsrc = ssrc
    },
    new TripleG3.P2P.Security.NoOpCipher());

using var sender = new RtpVideoSender(
    new RtpVideoSenderConfig
    {
        RemoteIp = remote.EndPoint.Address.ToString(),
        RemotePort = remote.EndPoint.Port,
        Ssrc = ssrc
    },
    new TripleG3.P2P.Security.NoOpCipher());

receiver.FrameReceived += frame =>
{
    if (frame is not { } received) return;
    Console.WriteLine(received.AnnexB.Length);
    received.Dispose();
};

await receiver.StartAsync();

if (args.Length == 2)
{
    var annexB = await File.ReadAllBytesAsync(args[1]);
    using var frame = EncodedAccessUnit.FromAnnexB(annexB, DateTimeOffset.UtcNow.Ticks, true);
    await sender.SendAsync(frame);
}

await Task.Delay(Timeout.InfiniteTimeSpan);

Host-facing transport APIs

TripleG3.P2P exposes transport-only primitives that a host such as Alice can map into its own authorization, workflow, and UI state services:

• TripleG3.P2P.FileTransfer.IFileTransferClient provides explicit-consent, SHA-256-verified TCP file transfer. A receiver rejects offers by default if it has no TransferRequested handler.
• TripleG3.P2P.Core.ISubscriptionSerialBus.Subscribe<T> returns a disposable subscription. All registrations are cleared when the bus connection closes or the bus is disposed.
• TripleG3.P2P.Video.RtpVideoSender and RtpVideoReceiver provide H.264 Annex-B/RTP transport.
• TripleG3.P2P.Audio.RtpAudioSender and RtpAudioReceiver provide 48 kHz mono, 20 ms Opus/RTP transport with bounded receiver delivery.
• IPeerAuthorizer is host-supplied and is evaluated before inbound serial, file-transfer, and audio media payloads are processed. Use an approved session/device allowlist; endpoint identity alone is not sufficient.
• File and audio transports expose P2PDiagnosticEventArgs for lifecycle, authorization, and transfer progress state projection.

Production callers must supply an authenticated encryption/channel implementation before accepting untrusted user traffic. NoOpCipher and test ciphers are not appropriate for deployed traffic.

Connected device hub

ConnectedDeviceHub<TDeviceDescriptor, TConnectionRoute, TStreamDescriptor> is a generic in-memory connection and routing projection. It does not open sockets or publish messages itself. Hosts add or remove already-trusted device connections, query immutable membership snapshots, and publish the returned dispatch plans using their selected transport.

The hub provides:

• stable device IDs paired with replaceable connection IDs;
• revisioned join, graceful-leave, and disconnect changes;
• current-device queries and stale-route detection through revocation tokens;
• direct and all-device routing for opaque host messages;
• transport-neutral live-session offer, answer, start, stop, and failure control;
• automatic live-session failure when a participating connection leaves or is replaced.

The host owns authentication, authorization, approval, persistence, request and notification queues, tool execution, media capture and rendering, network publication, and the live-session data plane.

High-performance, attribute-driven peer-to-peer messaging for .NET 10 / MAUI apps over UDP and TCP. Ship strongly-typed messages (records / classes / primitives / strings) with a tiny 8-byte header and pluggable serialization strategy.

Status: UDP and TCP transports plus three serializers (None, JsonRaw, and LengthPrefixed) are implemented. Direct peer-to-peer TCP file transfer is available. RTP/H.264 video remains experimental.

---

Why?

Typical networking layers force you to hand-roll framing, routing, and serialization. TripleG3.P2P gives you:

• A single minimal interface: ISerialBus (send, subscribe, start, close)
• Deterministic wire contract via [P2PProperty] & [P2PMessage] attributes (order + protocol name stability)
• Envelope-based dispatch that is assembly agnostic (type names / attribute names, not CLR identity)
• Choice between ultra-light delimiter serialization or raw JSON
• Safe, isolated subscriptions (late subscribers don't crash the loop)
• Zero allocations for header parsing (Span/Memory friendly design internally)
• NEW: Multi-endpoint fan-out with duplicate endpoint suppression

---

Features At A Glance

• Target framework: net10.0 (single TFM here; platform heads handled by MAUI host project)
• 8‑byte header layout (Length / MessageType / SerializationProtocol)
• Attribute ordered property serialization with stable delimiter @-@
• Automatic Envelope wrapping so handlers receive strong types directly
• Multiple simultaneous protocol instances via separate buses
• Multi-endpoint fan-out (one SendAsync → N peers)
• Plug-in serializer model (IMessageSerializer)
• Graceful cancellation / disposal
• TCP transport (reliable stream) alongside UDP; same API
• Direct peer-to-peer TCP file transfer with receiver consent, cancellation, progress, and SHA-256 verification
• Separate stdio MCP server for AI-assisted configuration, code generation, documentation, and troubleshooting

Transport Summary

Transport	Reliability	Ordering	Fan-Out	Status
UDP	Best effort / loss possible	Not guaranteed across datagrams	Yes (multi-endpoint send)	Implemented
TCP	Reliable	Preserved per connection	Yes (writes to each stream)	Implemented

Use UDP when you need lowest overhead and can tolerate loss; use TCP when you need reliability / ordering without building it yourself. Neither transport provides authentication or NAT traversal; applications must provide peer authorization and network connectivity.

Peer-to-peer file transfer

File transfer is intentionally separate from ISerialBus. Create a PeerFileTransferClient for each peer, start its listener, and handle incoming requests explicitly. A receiver can reject a request or accept it with a destination path; cancellation tokens cancel active sender or receiver operations. Transfers use a dedicated versioned TCP protocol, stream in bounded chunks, write to a temporary .part file, and verify SHA-256 before moving the completed file into place.

using TripleG3.P2P.FileTransfer;

var receiver = new PeerFileTransferClient(new FileTransferOptions {
    LocalEndPoint = new IPEndPoint(IPAddress.Loopback, 9100)
});
receiver.TransferRequested += (request, cancellationToken) =>
    new ValueTask<FileTransferDecision>(
        FileTransferDecision.Accept(Path.Combine("received", request.FileName)));
await receiver.StartAsync();

var sender = new PeerFileTransferClient(new FileTransferOptions {
    LocalEndPoint = new IPEndPoint(IPAddress.Loopback, 9101)
});
await sender.StartAsync();
var results = await sender.SendAsync(
    "video.mp4",
    [new IPEndPoint(IPAddress.Loopback, 9100)],
    cancellationToken: cancellationToken);

If no TransferRequested handler is registered, inbound transfers are rejected. The receiver may reject a request from the handler, and the sender may cancel through its cancellation token. Configure MaximumFileBytes, BufferSize, and MaximumConcurrentTransfers for the host.

---

RTP Audio

RtpAudioSender sends pre-encoded 48 kHz mono, 20 ms Opus frames to one configured RTP endpoint. RtpAudioReceiver binds a local endpoint, validates the payload type and configured SSRC, and raises AudioFrameReceived. The host supplies Opus encoding and decoding, authorization, NAT traversal, and production media security.

---

Hubs

The TripleG3.P2P.Hubs namespace provides authoritative in-memory state and routing policy above the transport layer. Hubs assign server timestamps and identifiers, expose immutable snapshots, maintain bounded message and notification histories, and return recipient member IDs for accepted chat or audio routes. Applications map member IDs to authenticated sessions and network endpoints.

Chat hubs

ChatHub is ownerless: zero or more members can join or leave themselves. HostedChatHub starts with one host; hosts can add, remove, promote, and demote members, while the final host cannot leave members behind.

var catalog = new HubCatalog();
var room = catalog.CreateChatHub(Guid.NewGuid());
var alice = Guid.NewGuid();
var bob = Guid.NewGuid();

room.Join(alice, "Alice");
room.Join(bob, "Bob");

HubDispatch dispatch = room.SendMessage(alice, "Hello");

For hosted moderation:

var host = Guid.NewGuid();
var room = catalog.CreateHostedChatHub(Guid.NewGuid(), host, "Host");
var player = Guid.NewGuid();

room.AddMember(host, player, "Player");
room.PromoteMember(host, player);
room.RemoveMember(player, host);

Gaming lobby hub

GamingLobbyHub adds zero or more teams, exclusive team assignment, all-lobby or team-only chat, and all-lobby or team-only RTP-audio routing policy. It returns a HubAudioRoute; the host maps those recipient IDs to dedicated RtpAudioSender instances.

var host = Guid.NewGuid();
var lobby = catalog.CreateGamingLobby(Guid.NewGuid(), host, "Host");
var red = Guid.NewGuid();
var player = Guid.NewGuid();

lobby.AddTeam(host, red, "Red");
lobby.AddMember(host, player, "Player");
lobby.AssignMemberToTeam(host, player, red);

HubDispatch chat = lobby.SendChat(player, HubAudience.Team, red, "Ready");
HubAudioRoute audio = lobby.GetAudioRoute(player, HubAudience.Team, red);

Custom hub messages

The existing hubs can route any user-defined payload type without creating a generic hub class. The hub supplies authoritative sender, audience, team, timestamp, revision, recipients, and hub ID in HubDispatch<TMessage>; the custom payload cannot override those values.

[P2PMessage("PlayerReady")]
public sealed record PlayerReady(
    [property: P2PProperty(1)] bool IsReady);

HubDispatch<PlayerReady> dispatch = lobby.RouteMessage(
    player,
    HubAudience.Team,
    red,
    new PlayerReady(true));

Ownerless and hosted chat hubs provide the simpler all-room overload:

HubDispatch<TypingIndicator> dispatch = room.RouteMessage(
    alice,
    new TypingIndicator(true));

Custom payloads are not retained in chat history by default. Applications may selectively persist them, then publish dispatch.Message to the authenticated sessions identified by dispatch.RecipientMemberIds. Payload types should use [P2PMessage] and [P2PProperty] when sent through ISerialBus.

Notifications hub

NotificationsHub registers local device profiles and routes a full platform-neutral notification to zero or more users, devices, or platforms. Every NotificationDelivery retains the full NotificationMessage and includes a typed Windows, Android, or iOS projection. It does not call Firebase, Apple Push Notification service, Windows App SDK, or Azure Notification Hubs.

var notifications = catalog.CreateNotificationsHub(Guid.NewGuid());
var device = notifications.RegisterDevice(
    Guid.NewGuid(),
    alice,
    NotificationPlatform.Android,
    "en-US");

NotificationDispatch dispatch = notifications.Route(
    new NotificationRequest(
        "Match ready",
        "Open the game.",
        Category: "game",
        Data:
        [
            new NotificationDataEntry("androidChannelId", "matches"),
            new NotificationDataEntry("androidSmallIcon", "ic_match")
        ]),
    NotificationRecipient.ForDevices(device.DeviceId));

For delivery through ISerialBus, convert each result to NotificationWireDelivery. Receiving code can read either representation:

NotificationWireDelivery wire = dispatch.Deliveries[0].ToWireDelivery();
NotificationMessage full = wire.ReadNotification();
AndroidNotificationView android = wire.ReadPlatformView<AndroidNotificationView>();

The platform views are application-facing data models only. Device code decides how to display them with the relevant operating-system APIs.

Only trusted host code should register or remove devices and create recipient selectors. The host must authorize user and device identifiers before invoking NotificationsHub; payload-provided identities are not trusted.

Video chat hub

VideoChatHub is an ownerless zero-to-many participant room for chat, custom signaling, and multiparty media routing. Camera and microphone states are independent and disabled by default. The hub returns recipient IDs and validates route freshness; the host owns capture, encoding, encryption, RTP sender/receiver instances, decoding, rendering, and playback.

var videoChat = catalog.CreateVideoChatHub(Guid.NewGuid());
var alice = Guid.NewGuid();
var bob = Guid.NewGuid();

videoChat.Join(alice, "Alice");
videoChat.Join(bob, "Bob");
videoChat.SetCameraEnabled(alice, true);
videoChat.SetMicrophoneEnabled(alice, true);

VideoChatRecipientRoute mediaRoute = videoChat.GetMediaRoute(
    alice,
    VideoChatMediaKind.AudioAndVideo);

Use one monotonic capture timestamp to derive corresponding RTP clock values:

var captureOrigin = Stopwatch.GetTimestamp();
var clock = new RtpMediaClock(captureOrigin, Stopwatch.Frequency);
var captureTimestamp = Stopwatch.GetTimestamp();
RtpMediaTimestamps timestamps = clock.Map(captureTimestamp);

var audioMetadata = new AudioFrameMetadata(timestamps.AudioTimestamp48k);
using var videoFrame = new EncodedAccessUnit(
    annexB,
    isKeyFrame,
    timestamps.VideoTimestamp90k,
    captureTimestamp);

Check IsRouteCurrent and RevocationToken during each fan-out batch. Membership and camera/microphone changes invalidate and cancel prior media routes; text and custom signaling do not. RTP timestamps share one publisher's monotonic capture origin. This is timestamp correspondence, not receiver playout synchronization. Production synchronization across devices still requires authenticated signaling and RTCP clock correlation.

Register the process-wide catalog with dependency injection:

services.AddP2PHubs();
var catalog = serviceProvider.GetRequiredService<IHubCatalog>();

Hubs do not trust member IDs received from payloads and do not publish directly to network transports. All requester and member ID parameters must be trusted identities resolved by the hosting application from an authenticated session. The host then publishes HubDispatch or applies HubAudioRoute. Check GamingLobbyHub.IsAudioRouteCurrent immediately before applying an audio route; membership, team, and policy changes invalidate prior revisions. Do not implement team delivery by temporarily mutating one bus's global outbound endpoint set; use targeted session delivery or one stable transport per audience.

---

Installation

Install the core library from NuGet:

dotnet add package TripleG3.P2P

Symbols are published; enable source stepping to debug internals.

The MCP server is a separate executable and is not included in the NuGet package. Run it locally from the repository with:

dotnet run --project src/TripleG3.P2P.McpServer/TripleG3.P2P.McpServer.csproj --no-launch-profile

VS Code integration is configured in .vscode/mcp.json.

Versioning & CI

The release workflow derives Major.Minor.RunNumber from the project version and publishes the package when a push reaches main. Set the repository NUGET_API_KEY secret before enabling publishing. Local packages can be created with dotnet pack src/TripleG3.P2P/TripleG3.P2P.csproj -c Release.

---

Target Framework

net10.0

MAUI/platform variants are produced in a sibling project; this core library stays lean.

---

Core Concepts

ISerialBus

public interface ISerialBus {
    bool IsListening { get; }
    ValueTask StartListeningAsync(ProtocolConfiguration config, CancellationToken ct = default);
    ValueTask CloseConnectionAsync();
    void SubscribeTo<T>(Action<T> handler);
    ValueTask SendAsync<T>(T message, MessageType messageType = MessageType.Data, CancellationToken ct = default);
}

Abstracts the transport (currently UDP + TCP). Your code remains identical besides construction via the factory.

Both built-in buses also implement ISubscriptionSerialBus, whose Subscribe<T> method returns an IDisposable unsubscription registration.

ProtocolConfiguration

public sealed class ProtocolConfiguration {
    IPAddress LocalAddress { get; init; }
    IReadOnlyCollection<IPEndPoint> OutboundEndPoints { get; init; }
    int        LocalPort      { get; init; }
    int        MaxPayloadBytes { get; init; }
    int        MaxInboundConnections { get; init; }
    SerializationProtocol SerializationProtocol { get; init; }
}

Controls binding, outbound destinations, and the serialization protocol used for every message on this bus instance.

Outbound Destinations / Fan-Out

Configure OutboundEndPoints to fan out every SendAsync from a bus instance. The collection has set semantics: duplicate endpoints with the same address:port are suppressed. An empty collection creates a receive-only bus; SendAsync throws until at least one outbound endpoint is configured. These are configured unicast peers, not IP broadcast addresses.

Local Development: Multiple Processes on One Host

When all peers run on the same computer, each listener needs a different loopback port. This configuration sends from the local hub at 127.0.0.1:7000 to three local receiver processes:

await bus.StartListeningAsync(new ProtocolConfiguration {
    LocalAddress = IPAddress.Loopback,
    LocalPort = 7000,
    OutboundEndPoints = [
        new IPEndPoint(IPAddress.Loopback, 7001),
        new IPEndPoint(IPAddress.Loopback, 7002),
        new IPEndPoint(IPAddress.Loopback, 7003)
    ],
    SerializationProtocol = SerializationProtocol.None
});

await bus.SendAsync(new Chat("me", "hi everyone")); // reaches 7001, 7002, 7003

Configure the three receiver processes with LocalAddress = IPAddress.Loopback and local ports 7001, 7002, and 7003. They cannot all bind 127.0.0.1:7000 on the same host.

Remote Devices: Shared Service Port Where Available

Remote devices have unique IP addresses, so they can usually all listen on the same service port. This hypothetical hub at 10.42.0.10:7000 fans out to two devices on port 7000 and a third on port 7100 because that device already has another local service using 7000:

await hub.StartListeningAsync(new ProtocolConfiguration {
    LocalAddress = IPAddress.Parse("10.42.0.10"),
    LocalPort = 7000,
    OutboundEndPoints = [
        new IPEndPoint(IPAddress.Parse("10.42.0.21"), 7000),
        new IPEndPoint(IPAddress.Parse("10.42.0.22"), 7000),
        new IPEndPoint(IPAddress.Parse("10.42.0.23"), 7100) // Port 7000 is unavailable on this device.
    ],
    SerializationProtocol = SerializationProtocol.LengthPrefixed
});

await hub.SendAsync(new Announcement("server", "hello remote peers"));

The receiving devices bind these local endpoints:

Device	Local endpoint
Device A	10.42.0.21:7000
Device B	10.42.0.22:7000
Device C	10.42.0.23:7100

Each device can use the same LocalPort when its IP address is different. Configure LocalAddress with the device's actual interface address and add a route, firewall rule, or NAT mapping as needed; this library does not provide NAT traversal.

Dynamic Lobby Membership

ProtocolConfiguration.OutboundEndPoints seeds the bus at startup. For a running UDP or TCP bus, cast it to IOutboundEndpointSerialBus to change the destination set without restarting its listener:

var endpointBus = (IOutboundEndpointSerialBus)bus;

endpointBus.AddOutboundEndPoint(joiningPeer);
endpointBus.RemoveOutboundEndPoint(leavingPeer);

Changes apply to future SendAsync calls. Removing a TCP endpoint also closes its outbound connection; a send that was already in progress may still finish.

Full Mesh (N peers each send to all others): create N buses where each bus configures every other peer in OutboundEndPoints. (See integration test Concurrent_FanOuts_All_Messages_Delivered_Exactly_Once).

Duplicate endpoint suppression:

await bus.StartListeningAsync(new ProtocolConfiguration {
    LocalPort = 8100,
    OutboundEndPoints = [
        new IPEndPoint(IPAddress.Loopback, 8101),
        new IPEndPoint(IPAddress.Loopback, 8102),
        new IPEndPoint(IPAddress.Loopback, 8102)
    ]
});
// Only one datagram sent to 8102.

Mixed types fan-out (string + complex):

await sender.SendAsync("alpha");
await sender.SendAsync(new Person("carol", 27, new Address("2 Ave", "Metro")));

Concurrency: safe to invoke SendAsync concurrently; the underlying UDP socket will serialize sends. Out-of-order arrival is still possible (UDP). If ordering matters, include sequence numbers inside your messages.

Reliability Disclaimer: UDP does not guarantee delivery / ordering. The library only guarantees attempted one-shot fan-out; implement retries / ACK at the message layer if required.

Dependency Injection

Register UDP components:

services.AddP2PUdp(); // Registers None + JsonRaw + LengthPrefixed serializers and UDP bus
var bus = services.BuildServiceProvider().GetRequiredService<ISerialBus>();

Need TCP as well? Use SerialBusFactory.CreateTcp(). A combined TCP DI extension is not currently provided.

Envelope (Generic)

Internal transport wrapper: TypeName + Message. The receiver inspects TypeName to look up subscriptions, then materializes only the requested type.

SerializationProtocol

None    // Attribute-delimited (fast, compact)
JsonRaw // System.Text.Json UTF-8 payload
LengthPrefixed // Versioned attribute contract with lengths and explicit null markers

Add more by implementing IMessageSerializer.

Use LengthPrefixed for new attribute-based contracts. None remains available for wire compatibility, but delimiter text inside values, null versus empty strings, and some nested shapes are inherently ambiguous.

Attributes

• [P2PMessage] or [P2PMessage("CustomName")] gives the logical protocol name (stable across assemblies)
• [P2PMessage<T>] generic variant uses typeof(T).Name (or supplied override) for convenience
• [P2PProperty(order)] marks and orders properties participating in attribute serialization.
• Unannotated properties are ignored by None and LengthPrefixed.
• Constructor parameters are matched to annotated properties by name and type.

P2PMessageAttribute replaces UdpMessageAttribute, and P2PPropertyAttribute replaces UdpAttribute. Change [UdpMessage] to [P2PMessage] and [Udp(order)] to [P2PProperty(order)] in existing contracts; protocol-visible names, ordering, and wire formats are unchanged.

MessageType

Currently: Data (extensible placeholder for control, ack, etc.)

---

Wire Format (UDP)

Header (8 bytes total):

1. Bytes 0-3: Int32 PayloadLength (bytes after header)
2. Bytes 4-5: Int16 MessageType
3. Bytes 6-7: Int16 SerializationProtocol

Payload:

• If SerializationProtocol.None: TypeName + optional @-@ + serialized property segments (each delimited by @-@)
• If JsonRaw: UTF-8 JSON of the Envelope<T>
• If LengthPrefixed: format version + length-prefixed UTF-8 type name + recursively framed ordered properties

---

Quick Start

using TripleG3.P2P.Attributes;
using TripleG3.P2P.Core;
using System.Net;

[P2PMessage("Person")] // Protocol type name
public record Person([property: P2PProperty(1)] string Name,
                     [property: P2PProperty(2)] int Age,
                     [property: P2PProperty(3)] Address Address);

[P2PMessage<Address>] // Uses nameof(Address) unless overridden
public record Address([property: P2PProperty(1)] string Street,
                      [property: P2PProperty(2)] string City,
                      [property: P2PProperty(3)] string State,
                      [property: P2PProperty(4)] string Zip);

var bus = SerialBusFactory.CreateUdp();
await bus.StartListeningAsync(new ProtocolConfiguration {
    LocalPort = 7000,
    OutboundEndPoints = [new IPEndPoint(IPAddress.Loopback, 7001)],
    SerializationProtocol = SerializationProtocol.None
});

bus.SubscribeTo<Person>(p => Console.WriteLine($"Person: {p.Name} ({p.Age}) {p.Address.City}"));

await bus.SendAsync(new Person("Alice", 28, new Address("1 Way", "Town", "ST", "00001")));

Run a second process with reversed ports (7001 ↔ 7000) to complete the loop.

TCP Quick Start (Reliable)

var tcpBus = SerialBusFactory.CreateTcp();
await tcpBus.StartListeningAsync(new ProtocolConfiguration {
    LocalPort = 9000,
    OutboundEndPoints = [new IPEndPoint(IPAddress.Loopback, 9001)],
    SerializationProtocol = SerializationProtocol.JsonRaw
});

tcpBus.SubscribeTo<Person>(p => Console.WriteLine($"[TCP] {p.Name} ({p.Age})"));
await tcpBus.SendAsync(new Person("Alice", 28, new Address("1 Way", "Town", "ST", "00001")));

Multi-Protocol Usage (Same Message Types)

var udp = SerialBusFactory.CreateUdp();
var tcp = SerialBusFactory.CreateTcp();
await udp.StartListeningAsync(new ProtocolConfiguration { LocalPort = 7000, OutboundEndPoints = [new IPEndPoint(IPAddress.Loopback, 7001)], SerializationProtocol = SerializationProtocol.None });
await tcp.StartListeningAsync(new ProtocolConfiguration { LocalPort = 9000, OutboundEndPoints = [new IPEndPoint(IPAddress.Loopback, 9001)], SerializationProtocol = SerializationProtocol.JsonRaw });

udp.SubscribeTo<Chat>(c => Console.WriteLine($"[UDP] {c.User}: {c.Text}"));
tcp.SubscribeTo<Chat>(c => Console.WriteLine($"[TCP] {c.User}: {c.Text}"));

await udp.SendAsync(new Chat("me","low-latency"));
await tcp.SendAsync(new Chat("me","reliable"));

---

Using JSON Instead

var jsonBus = SerialBusFactory.CreateUdp();
await jsonBus.StartListeningAsync(new ProtocolConfiguration {
    LocalPort = 7002,
    OutboundEndPoints = [new IPEndPoint(IPAddress.Loopback, 7003)],
    SerializationProtocol = SerializationProtocol.JsonRaw
});

JSON serializer ignores [P2PProperty] ordering—standard JSON rules apply; TypeName embedded as typeName/TypeName.

---

Subscriptions

bus.SubscribeTo<string>(s => Console.WriteLine($"Raw string: {s}"));
bus.SubscribeTo<Person>(HandlePerson);

void HandlePerson(Person p) { /*...*/ }

• Multiple handlers per type allowed
• Subscription key is the protocol type name (attribute override or CLR name)
• If no handler matches, message is silently ignored
• Cast to ISubscriptionSerialBus and use Subscribe<T> when the handler needs an independent lifetime.

---

Sending

await bus.SendAsync("Hello peer");
await bus.SendAsync(new Person("Bob", 42, new Address("2 Road", "City", "ST", "22222")));

All messages on a bus instance use that instance’s SerializationProtocol.

---

Graceful Shutdown

await bus.CloseConnectionAsync();
// or dispose
(bus as IDisposable)?.Dispose();

Cancels the receive loop & disposes socket.

---

Designing Message Contracts (Delimited Serializer)

1. Add [P2PMessage] (optional if CLR name is acceptable) to each root message type.
2. Annotate properties you want serialized with [P2PProperty(order)] (1-based ordering recommended).
3. Use only deterministic, immutable shapes (records ideal).
4. Nested complex types must also follow the same attribute pattern.
5. Changing order or adding/removing annotated properties is a protocol breaking change.

Example

[P2PMessage("Ping")] public record Ping([property: P2PProperty(1)] long Ticks);

Primitive & String Support

Primitive-like types use invariant round-trip formatting and explicit converters, including DateTimeOffset.

---

Implementing a Custom Serializer

class MyBinarySerializer : IMessageSerializer {
    public SerializationProtocol Protocol => (SerializationProtocol)42; // Add new enum value first
    public byte[] Serialize<T>(T value) { /* return bytes */ }
    public T? Deserialize<T>(ReadOnlySpan<byte> data) { /* parse */ }
    public object? Deserialize(Type t, ReadOnlySpan<byte> data) { /* parse */ }
}

Registration options:

1. Extend SerialBusFactory with a helper that injects your serializer.
2. Or (DI) register it as another IMessageSerializer; the bus chooses by Protocol enum value.

Best practices:

• Keep format deterministic & version tolerant.
• Reuse buffers; avoid per-message large allocations.
• Reserve new enum value before shipping (ensure both sides understand it).

---

Error Handling & Resilience

• Malformed, truncated, oversized, or unknown-protocol frames are rejected before deserialization or large allocation.
• Cancellation is propagated to callers.
• A send throws when every configured endpoint fails; partial fan-out failures are logged.
• Individual subscriber exceptions are logged and do not block other handlers.
• TCP serializes writes per configured connection and removes failed connections so a later send can reconnect.

---

Performance Notes

• Header parsing uses BinaryPrimitives on a single span
• Delimited serializer caches reflection lookups per type
• No dynamic allocations for header path; serialization aims to minimize intermediate copies
• Envelope design avoids repeated type discovery; only TypeName string extracted first

---

Extending To Other Transports

Transport abstraction lives behind ISerialBus.

TCP (Implemented)

Use SerialBusFactory.CreateTcp() and the same ProtocolConfiguration (LocalPort is the listener; OutboundEndPoints are configured send targets). Accepted sockets are receive-only sessions and are never added to fan-out. Ordering is preserved per configured connection, while different peers may progress independently.

Example:

var tcpBus = SerialBusFactory.CreateTcp();
await tcpBus.StartListeningAsync(new ProtocolConfiguration {
    LocalPort = 9000,
    OutboundEndPoints = [new IPEndPoint(IPAddress.Loopback, 9001), new IPEndPoint(IPAddress.Loopback, 9002)],
    SerializationProtocol = SerializationProtocol.JsonRaw
});
tcpBus.SubscribeTo<Chat>(c => Console.WriteLine($"[TCP] {c.User}: {c.Text}"));
await tcpBus.SendAsync(new Chat("alice", "over tcp"));

---

Examples and Tests

Tests are split by execution contract:

• TripleG3.P2P.UnitTests contains deterministic, in-process serializer, packetizer, depacketizer, cipher, sequence, bounds, and validation tests.
• TripleG3.P2P.IntegrationTests contains loopback sockets, multi-peer fan-out, reconnect, malformed network input, receiver lifecycle, DI-over-UDP video, RTCP timing, and negotiation flows.

The pull-request pipeline restores, builds, and runs only the unit-test project for fast feedback. The main-branch release pipeline also runs integration tests, validates the packed package, and builds a clean package consumer. Integration tests bind local ports and use timing-sensitive multi-component flows, so run the full solution locally before a release.

# Pipeline-equivalent unit tests
dotnet test tests/TripleG3.P2P.UnitTests/TripleG3.P2P.UnitTests.csproj -c Release -warnaserror

# Manual integration tests
dotnet test tests/TripleG3.P2P.IntegrationTests/TripleG3.P2P.IntegrationTests.csproj -c Release -warnaserror

# Complete local validation
dotnet test TripleG3.P2P.slnx -c Release -warnaserror

The solution also contains the executable MCP server project. It is built by the solution but is not packed into TripleG3.P2P.

The integration suite (MultiFanOutTests, TcpIntegrationTests, TransportHardeningTests, and the video integration fixtures) proves:

• UDP multi-endpoint fan-out
• TCP fan-out & ordering guarantees
• Concurrent full-mesh delivery
• Duplicate endpoint de-duplication
• TCP reconnect and concurrent frame integrity
• Malformed input recovery, cancellation, and disposable subscriptions
• All three serialization protocols
• RTP video DI transfer over a real UDP socket
• Ownerless and hosted chat routing over real UDP and TCP sockets
• Gaming all-lobby and team-only chat isolation over real UDP and TCP sockets
• Gaming team-only RTP audio delivery and stale route invalidation
• Notification user/device/platform routing and full/platform-specific parsing over UDP and TCP
• Video chat text/custom signaling over UDP/TCP and common-origin RTP audio/video timestamp delivery
• Receiver start/stop/restart and disposal races
• RTCP timing and negotiation/keyframe signaling

---

Roadmap

• Authentication and authorization for messaging and file-transfer peers
• Resumable file transfers and receiver-selected progress reporting
• Optional authenticated secure-channel wrapper
• Source generator for zero-reflection fast path
• Optional compression & encryption layers
• Health / metrics callbacks

---

FAQ

Q: Do both peers need the exact same CLR types?
A: They need matching protocol type names and compatible property ordering (Delimited) or matching JSON contracts (JsonRaw). CLR assembly identity is not required.

Q: Can I mix serializers on the same socket?
A: One ISerialBus instance uses one SerializationProtocol. Create multiple instances for mixed protocols.

Q: Is ordering enforced?
A: Receiver trusts the order defined by [P2PProperty(n)]. Reordering is a breaking change.

---

Minimal Cheat Sheet

var bus = SerialBusFactory.CreateUdp();
await bus.StartListeningAsync(new ProtocolConfiguration {
    LocalPort = 7000,
    OutboundEndPoints = [new IPEndPoint(IPAddress.Loopback, 7001)],
    SerializationProtocol = SerializationProtocol.None
});

[P2PMessage("Chat")]
public record Chat([property: P2PProperty(1)] string User, [property: P2PProperty(2)] string Text);

bus.SubscribeTo<Chat>(c => Console.WriteLine($"{c.User}: {c.Text}"));
await bus.SendAsync(new Chat("me", "hi there"));

---

License

GPL-3.0-only. See LICENSE.

---

Contributing

Issues & PRs welcome: add tests / samples for new serializers or transports.

---

Happy messaging!

---

Memory Pooling (Partially Implemented)

The video pipeline now uses pooled buffers for:

• FU‑A (fragmented) NAL reassembly (rents a buffer, grows, returns after NAL complete)
• Consolidated frame (Annex B access unit) assembly via an ArrayPoolFrame inside EncodedAccessUnit (returned when disposed)
• Intermediate single‑NAL copies (rented arrays tracked and released after frame consolidation)

Why: Reduce allocation pressure & GC pauses under sustained 30/60fps streaming. Hundreds or even thousands of frame buffers per minute are efficiently recycled.

Still Allocating:

• Outbound RTP packets use exact-length managed arrays so callback consumers may retain them safely.
• Control / negotiation JSON payloads

Guidance: Dispose each EncodedAccessUnit you create/receive (tests show using var au = ...). If you skip disposal, large buffers remain rented and the pool cannot reclaim them quickly.

Planned refinements: size threshold to bypass pooling for very small frames, packet buffer recycling, allocation metrics counters.

---

Experimental: Real-Time Video (H.264 over RTP)

Early, evolving support for sending pre‑encoded H.264 access units (Annex B) over RTP/UDP.

Current Capabilities

• H.264 packetization: single NAL + FU‑A (RFC 6184 subset)
• Depacketization & reassembly to Annex B (start codes preserved)
• Minimal RTCP: Sender Report (SR) + Receiver Report (RR) with RTT & fraction lost
• Jitter, cumulative loss, fraction lost statistics (RFC3550 style)
• Forward‑only reorder buffer (drops late retrograde packets)
• Keyframe request path (PLI analogue) via control channel
• Simple JSON negotiation (Offer/Answer + keyframe request) over injectable reliable channel
• Pluggable payload cipher abstraction (NoOp / XOR test cipher)
• Partial memory pooling & zero‑copy optimizations for frame assembly

Not Yet Implemented / Limitations

• No NACK/RTX retransmissions, FEC, or packet pacing
• No SRTP / DTLS keying (encryption is placeholder only)
• No automatic SPS/PPS extraction; callers may supply ProfileLevelId and SpropParameterSets during negotiation.
• No adaptive bitrate / congestion control (REMB/TCC/GCC)
• Reordering is bounded and loss-recovering, but NACK/RTX is not implemented.
• Only H.264 (no VP8/VP9/AV1/H.265)
• No simulcast / SVC layers, no audio, no A/V sync
• No ICE/STUN/TURN NAT traversal (you must provide transport)

Essential Types

Type	Purpose
EncodedAccessUnit	Represents one encoded frame (Annex B) + metadata; disposable (pooled buffer)
RtpVideoSender	High-level sender: packetizes AUs, encrypts, emits RTP datagrams & sends SR
RtpVideoReceiver	Consumes RTP / RTCP, reassembles frames, updates stats, raises AccessUnitReceived
H264RtpPacketizer / H264RtpDepacketizer	Low-level packetization building blocks
NegotiationManager	Offer/Answer & keyframe (PLI analogue) signaling
IVideoPayloadCipher	Cipher abstraction (NoOp / XOR examples)

Basic End‑to‑End Flow

// Outbound network send hooks (wire these to your UDP socket send method)
void SendRtp(ReadOnlySpan<byte> datagram) => udpSocket.SendTo(datagram.ToArray(), remoteRtpEndPoint);
void SendRtcp(ReadOnlySpan<byte> datagram) => udpSocket.SendTo(datagram.ToArray(), remoteRtcpEndPoint);

var sender = new RtpVideoSender(
    ssrc: 0x1234_5678,
    mtu: 1200,
    cipher: new NoOpCipher(),
    datagramOut: d => SendRtp(d.Span),
    rtcpOut: d => SendRtcp(d.Span));

var receiver = new RtpVideoReceiver(new NoOpCipher());
receiver.AccessUnitReceived += au => {
    try { Render(au); } finally { au.Dispose(); }
};

// For each encoded frame (Annex B) you obtain from your encoder:
using var au = new EncodedAccessUnit(encodedAnnexBFrame, isKeyFrame, rtpTimestamp90k, captureTicks);
sender.Send(au);

// Periodically (every ~2s or on key events) send a Sender Report:
sender.SendSenderReport(rtpTimestamp90k);

// Incoming network data:
void OnUdpData(byte[] buffer, int len)
{
    var span = new ReadOnlySpan<byte>(buffer,0,len);
    if (Rtcp.IsRtcpPacket(span))
    {
        receiver.ProcessRtcp(span);
        sender.ProcessRtcp(span); // so sender can compute RTT
    }
    else
    {
        receiver.ProcessRtp(span);
    }
}

Negotiation & Keyframe Requests

var chA = new InMemoryControlChannel();
var chB = new InMemoryControlChannel();
chA.MessageReceived += m => chB.SendReliableAsync(m);
chB.MessageReceived += m => chA.SendReliableAsync(m);

var offerer = new NegotiationManager(chA);
var answerer = new NegotiationManager(chB);
answerer.AttachEncoder(yourEncoder); // so remote PLI triggers RequestKeyFrame()

await offerer.CreateOfferAsync(new VideoSessionConfig(1280,720, 2_000_000, 30));
// After negotiation completes, either side can request a keyframe:
offerer.RequestKeyFrame();

Stats & RTT

var senderStats = sender.GetStats();
// senderStats.RttEstimateMs, PacketsSent, BytesSent
var recvStats = receiver.GetStats();
// recvStats.Jitter, PacketsLost, FractionLost (last interval), PacketsReceived

Memory & Disposal

Dispose every received EncodedAccessUnit after consuming its data. It returns the underlying pooled frame buffer to ArrayPool<byte>. If you retain frames (e.g., for rewind), copy out the bytes first with au.AnnexB.ToArray().

Keyframe Logic

Receiver issues RequestKeyFrame() on NegotiationManager (or directly via application logic) when:

• Startup / first frame needed
• Decoder error / corruption detected (you decide)

The answering side invokes encoder.RequestKeyFrame() (you supply the encoder implementation) so the next access unit is a keyframe.

Roadmap Snapshot

Minimal RTP Video API (Stable Surface)

Verbatim stable signatures exposed in TripleG3.P2P.Video namespace:

public sealed class EncodedAccessUnit : IDisposable
{
    public EncodedAccessUnit(ReadOnlyMemory<byte> annexB, bool isKeyFrame, uint rtpTimestamp90k, long captureTicks);
    public ReadOnlyMemory<byte> AnnexB { get; }
    public bool IsKeyFrame { get; }
    public uint RtpTimestamp90k { get; }
    public long CaptureTicks { get; }
    public void Dispose();
}

public interface IVideoPayloadCipher
{
    int OverheadBytes { get; }
    int Encrypt(Span<byte> buffer);
    int Decrypt(Span<byte> buffer);
    int Encrypt(ReadOnlySpan<byte> payload, Span<byte> output);
    int Decrypt(ReadOnlySpan<byte> payload, Span<byte> output);
}

public sealed class NoOpCipher : IVideoPayloadCipher { /* OverheadBytes=0; pass-through */ }

public sealed class RtpVideoSender
{
    public RtpVideoSender(uint ssrc, int mtu, IVideoPayloadCipher cipher, Action<ReadOnlyMemory<byte>> datagramOut, Action<ReadOnlyMemory<byte>>? rtcpOut = null);
    public void Send(EncodedAccessUnit au);
    public void SendSenderReport(uint rtpTimestamp90k);
    public void ProcessRtcp(ReadOnlySpan<byte> packet);
    public RtpVideoSenderStats GetStats(); // optional lightweight stats
}

public sealed class RtpVideoReceiver
{
    public RtpVideoReceiver(IVideoPayloadCipher cipher);
    public event Action<EncodedAccessUnit> AccessUnitReceived;
    public void ProcessRtp(ReadOnlySpan<byte> packet);
    public void ProcessRtcp(ReadOnlySpan<byte> packet);
    public RtpVideoReceiverStats GetStats();
}

public static class Rtcp
{
    public static bool IsRtcpPacket(ReadOnlySpan<byte> packet);
}

These are intended for direct consumption by TripleG3.Camera.Maui without reflection.

Logging & Diagnostics

Add logging (video pipeline & new code paths use Microsoft.Extensions.Logging):

services.AddLogging(b => b.AddConsole().SetMinimumLevel(LogLevel.Information));

Security Note (Video Ciphers)

NoOpCipher & XorTestCipher are NOT secure. They exist only for testing. Use a proper SRTP / DTLS-SRTP layer for real encryption (on roadmap).

Creating a New Transport

1. Implement ISerialBus (mirror UDP/TCP structure).
2. Accept an IEnumerable<IMessageSerializer>.
3. Preserve the 8‑byte header (or version it explicitly).
4. Provide a SerialBusFactory.CreateX() helper.
5. Add integration tests: start, send, fan-out, mixed serializers.
6. Update README & bump minor version.

MAUI Integration

In MauiProgram.CreateMauiApp:

builder.Services.AddP2PUdp();
builder.Services.AddLogging(b => b.AddDebug());

Inject ISerialBus into pages / services. Handle disposal on shutdown for clean socket release.

• Immediate: NACK/RTX, proper RTCP PLI/FIR packets, SRTP integration
• Near-term: Bandwidth estimation (REMB/TCC) & adaptive send pacing
• Later: ICE/STUN/TURN integration hooks, multi‑codec negotiation, richer jitter buffer

APIs are experimental; expect adjustments.