Custom IPC Transport

You can create your own custom IPC (inter-process communication) transport by implementing the IEventTransport interface. Though Lite.EventIpc natively supports TCP/IP, Named Pipes, and Memory Maps, the implementation may not suit your needs. As an example, you may want to use DBus on Linux, or include a custom encryption layer for TCP/IP transports.

The IEventTransport has 3 key methods that must be implemented for custom IPC event transportation handler.

Methods:

• StartListening<TEvent>(Action<TEvent> onEventReceived) - Initiates the custom IPC to start listening
• StopListening() - Stops the custom IPC from listening
• Send<TEvent>(TEvent eventData) - _Send or 'Publish' message to IPC listener

IEventTransport Interface

/// <summary>
///   Bi-directional event transport interface used for sending and receiving event packages, AKA: "envelopes".
/// </summary>
public interface IEventTransport
{
  /// <summary>
  ///   Send a message (request or response). If envelope.IsResponse == true and envelope.ReplyTo != null,
  ///   transport sends to the reply channel; otherwise to its configured request channel.
  /// </summary>
  /// <typeparam name="TEvent">Type of event data.</typeparam>
  /// <param name="eventData">Event data to send.</param>
  void Send<TEvent>(TEvent eventData);

  /// <summary>
  ///   Start listening for both requests and responses (as applicable).
  /// </summary>
  /// <typeparam name="TEvent">Type of event data.</typeparam>
  /// <param name="onEventReceived">Message handler.</param>
  void StartListening<TEvent>(Action<TEvent> onEventReceived);

  /// <summary>Stop listensing or responses operations.</summary>
  void StopListening();
}

Sample

The following is an implementation of the TCP/IP Transport

Implementation

using System.Net;
using System.Threading.Tasks;

namespace SampleApp;

// Data to transport
public record Ping(string Message);
public record Pong(string Message);

private const int RequestSendPort = 6001;
private readonly string _addrClient = IPAddress.Loopback.ToString();
private readonly string _addrServer = IPAddress.Loopback.ToString();
private string _msgPayload = "hello";
private bool _msgReceived = false;

public void SampleCustomTransport()
{
  var msgPayload = "hello";
  var msgReceived = false;

  var serverTransport = new CustomTcpTransport(_addrServer, RequestSendPort);
  var clientTransport = new CustomTcpTransport(_addrClient, RequestSendPort);

  // Server listener
  serverTransport.StartListening<Ping>(req =>
  {
    if (req.Message == msgPayload)
      msgReceived = true;
  });

  // Client sender
  clientTransport.Send(new Ping(msgPayload));

  // Give it a moment
  Task.Delay(Timeout50).Wait();
  Assert.IsTrue(msgReceived);
}

Custom Transport

using System;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Lite.EventIpc;
using Lite.EventIpc.IpcTransport;

namespace SampleApp;

public class CustomTcpTransport : IEventTransport
{
  private readonly string _host;
  private readonly int _port;
  private CancellationToken _cancelToken;
  private CancellationTokenSource? _cts;

  public TcpTransport(string host, int port)
  {
    _host = host;
    _port = port;
  }

  public void Send<TEvent>(TEvent eventData)
  {
    // Serialize and send over TCP socket
    var json = EventSerializer.Serialize(eventData);
    using var client = new TcpClient(_host, _port);

    var stream = client.GetStream();
    var bytes = Encoding.UTF8.GetBytes(json);

    stream.Write(bytes, 0, bytes.Length);
  }

  public void StartListening<TEvent>(Action<TEvent> onEventReceived)
  {
    _cts = new CancellationTokenSource();
    _cancelToken = _cts.Token;

    // Listen on TCP socket and deserialize
    Task.Run(() =>
    {
      var listener = new TcpListener(IPAddress.Any, _port);
      listener.Start();

      try
      {
        while (!_cancelToken.IsCancellationRequested)
        {
          using var client = listener.AcceptTcpClient();
          using var stream = client.GetStream();

          var buffer = new byte[4096];
          var bytesRead = stream.Read(buffer, 0, buffer.Length);

          if (bytesRead <= 0)
            continue;

          var json = Encoding.UTF8.GetString(buffer, 0, bytesRead);
          var evt = EventSerializer.Deserialize<TEvent>(json);

          onEventReceived(evt);
        }
      }
      catch (Exception)
      {
        /* Swallow exceptions until we get logging coming later */
      }
      finally
      {
        try
        {
          listener.Stop();
        }
        catch
        {
        }
      }
    });
  }

  /// <inheritdoc/>
  public void StopListening()
  {
    _cts?.Cancel();
  }
}