Start using PairSockets for two-way debug communication

This commit is contained in:
Yoshi Askharoun
2023-05-29 20:19:31 -05:00
parent a61e1e63a9
commit d01235e3f4
3 changed files with 35 additions and 21 deletions
+15
View File
@@ -0,0 +1,15 @@
using System.Runtime.Serialization;
namespace Microsoft.Iris.Debug;
/// <summary>
/// Provides helpers for things that are common between debug clients, servers, and transports.
/// </summary>
public static class DebugRemoting
{
public const string DEFAULT_TCP_CLIENT_URI = ">tcp://127.0.0.1:5555,@tcp://127.0.0.1:55556";
public const string DEFAULT_TCP_SERVER_URI = "@tcp://127.0.0.1:5555,>tcp://127.0.0.1:55556";
internal static IFormatter CreateBsonFormatter() => new BsonFormatter(new StreamingContext(StreamingContextStates.Remoting));
}
@@ -1,4 +1,5 @@
using Microsoft.Iris.Debug.Data;
using Microsoft.Iris.Debug;
using Microsoft.Iris.Debug.Data;
using NetMQ;
using NetMQ.Sockets;
using System;
@@ -7,12 +8,12 @@ using System.IO;
using System.Runtime.Serialization;
using System.Text;
namespace Microsoft.Iris.Debug;
namespace Microsoft.Iris.Debug.NetMQ;
public class ZmqDebuggerClient : IDebuggerClient, IDisposable
{
private List<byte[]> _frames = new(2);
private readonly SubscriberSocket _subSocket;
private readonly PairSocket _socket;
private readonly IFormatter _formatter;
public string ConnectionUri { get; }
@@ -22,23 +23,22 @@ public class ZmqDebuggerClient : IDebuggerClient, IDisposable
public ZmqDebuggerClient(string connectionUri)
{
ConnectionUri = connectionUri;
ConnectionUri = connectionUri ?? DebugRemoting.DEFAULT_TCP_CLIENT_URI;
_subSocket = new();
_subSocket.Connect(connectionUri);
_subSocket.SubscribeToAnyTopic();
_socket = new();
_socket.Connect(connectionUri);
_formatter = ZmqDebuggerServer.CreateFormatter();
_formatter = DebugRemoting.CreateBsonFormatter();
System.Threading.Thread th = new(MessageRecieveLoop);
th.Start();
}
public void Dispose() => _subSocket.Dispose();
public void Dispose() => _socket.Dispose();
private void MessageRecieveLoop()
{
while (!_subSocket.IsDisposed)
while (!_socket.IsDisposed)
{
DebuggerMessageType type;
byte[] bytes;
@@ -65,7 +65,7 @@ public class ZmqDebuggerClient : IDebuggerClient, IDisposable
private bool TryRecieveDebuggerMessage(out DebuggerMessageType type, out byte[] bytes)
{
if (!_subSocket.IsDisposed && _subSocket.TryReceiveMultipartBytes(ref _frames, 2))
if (!_socket.IsDisposed && _socket.TryReceiveMultipartBytes(ref _frames, 2))
{
type = (DebuggerMessageType)BitConverter.ToInt32(_frames[0], 0);
bytes = _frames[1];
@@ -76,4 +76,6 @@ public class ZmqDebuggerClient : IDebuggerClient, IDisposable
bytes = null;
return false;
}
internal static IFormatter CreateFormatter() => new BsonFormatter(new StreamingContext(StreamingContextStates.Remoting));
}
@@ -3,25 +3,24 @@ using NetMQ.Sockets;
using System;
using System.IO;
using System.Runtime.Serialization;
using System.Runtime.Serialization.Formatters.Binary;
using System.Text;
namespace Microsoft.Iris.Debug;
namespace Microsoft.Iris.Debug.NetMQ;
internal class ZmqDebuggerServer : IDebuggerServer, IDisposable
{
public static IDebuggerServer Current { get; private set; }
private readonly PublisherSocket _pubSocket;
private readonly PairSocket _socket;
private readonly byte[][] _messageFrame = new byte[2][];
private readonly IFormatter _formatter;
public ZmqDebuggerServer(string connectionUri)
{
_pubSocket = new();
_pubSocket.Bind(connectionUri);
_socket = new();
_socket.Bind(connectionUri ?? DebugRemoting.DEFAULT_TCP_SERVER_URI);
_formatter = CreateFormatter();
_formatter = DebugRemoting.CreateBsonFormatter();
Current = this;
}
@@ -39,7 +38,7 @@ internal class ZmqDebuggerServer : IDebuggerServer, IDisposable
SendDebuggerMessage(DebuggerMessageType.DispatcherStep, message);
}
public void Dispose() => _pubSocket.Dispose();
public void Dispose() => _socket.Dispose();
private void SendDebuggerMessage(DebuggerMessageType type, string message, Encoding encoding = null)
{
@@ -52,8 +51,6 @@ internal class ZmqDebuggerServer : IDebuggerServer, IDisposable
_messageFrame[0] = BitConverter.GetBytes((int)type);
_messageFrame[1] = bytes;
_pubSocket.SendMultipartBytes(_messageFrame);
_socket.SendMultipartBytes(_messageFrame);
}
internal static IFormatter CreateFormatter() => new BsonFormatter(new StreamingContext(StreamingContextStates.Remoting));
}