I would like constructive criticism of a networking plugin I've written for Unity 3D. The use-case of the code is for people in "creative coding," where simple TCP and UDP messages are used for communication with lighting controllers, micro-controllers, and other devices, in addition to other computers. The requirement was to make it totally non-blocking, even in the event of a delayed or failed TCP message, so that it would be suitable for use in live performances or guest-experiences in theme parks. It does not support persistent TCP.
A design choice was to separate sending and receiving into different classes, in a similar pattern to the graphical language Max. Each of these classes is wrapped by a Unity class (MonoBehaviour
derived), messages and errors being passed back and forth from the main Unity thread with ConcurrentQueue
.
TCP Sending
public class TcpThreadedClient
{
// For returning error messages to main
private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();
public void Send(string address, int port, string message)
{
var sendTask = new Task(() => SendTask(address, port, message));
sendTask.Start();
}
public void SendTask(string address, int port, string message)
{
using (TcpClient client = new TcpClient())
{
try
{
// Connect to endpoint.
Task connect = client.ConnectAsync(address, port);
connect.Wait();
// Convert the string data to byte data using ASCII encoding.
byte[] byteData = Encoding.ASCII.GetBytes(message);
// Get the NetworkStream and send,
// with NoDelay (disable caching).
client.NoDelay = true;
NetworkStream stream = client.GetStream();
stream.Write(byteData, 0, message.Length);
}
catch (SocketException se)
{
PrintError("SocketException connecting to " + address + ":" + port + ": " + se.ToString());
}
catch (Exception e)
{
PrintError("Unexpected exception connecting to " + address + ":" + port + ": " + e.ToString());
}
}
}
private void PrintError(string e)
{
errorQueue.Enqueue(e);
}
// Wrapperr uses this function to
// get error messages.
public bool GetError(out string outError)
{
// Out promises initialization
return errorQueue.TryDequeue(out outError);
}
TCP Receiving
// State object for reading client data asynchronously
public class TcpStateObject
{
// Client socket.
public Socket workSocket = null;
// Size of receive buffer.
public const int BufferSize = 1024;
// Receive buffer.
public byte[] buffer = new byte[BufferSize];
// Received data string.
public StringBuilder sb = new StringBuilder();
}
public class TcpThreadedServer
{
// Queue for sharing messages with Unity main thread.
private ConcurrentQueue<NetworkMessage> messageQueue = new ConcurrentQueue<NetworkMessage>();
// For returning error messages to main Unity thread
private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();
// Thread signal.
public static ManualResetEvent allDone = new ManualResetEvent(false);
// For reading new messages from Queue
// without exposing Queue to other classes.
public bool GetMessage(out NetworkMessage message)
{
if(messageQueue.TryDequeue(out message))
{
return true;
}
return false;
}
// Start the network listening thread.
public void StartListening(int port)
{
// uses Task, but with LongRunning option it will create
// new thread, not use one from ThreadPool.
var listenTask = new Task(() => ListenTask(port), TaskCreationOptions.LongRunning);
listenTask.Start();
}
private void ListenTask(int port)
{
// Establish the local endpoint for the socket.
IPAddress ipAddress = IPAddress.Any;
IPEndPoint localEndPoint = new IPEndPoint(ipAddress, port);
// Create a TCP/IP socket.
Socket listener = new Socket(ipAddress.AddressFamily,
SocketType.Stream, ProtocolType.Tcp);
// Bind the socket to the local endpoint and listen for incoming connections.
try
{
listener.Bind(localEndPoint);
listener.Listen(100);
while (true)
{
// Set the event to nonsignaled state.
allDone.Reset();
// Start an asynchronous socket to listen for connections.
listener.BeginAccept(
new AsyncCallback(AcceptCallback),
listener);
// Wait until a connection is made before continuing.
allDone.WaitOne();
}
}
catch (Exception e)
{
PrintError("Error opening TCP socket: " + e.ToString());
}
}
private void AcceptCallback(IAsyncResult ar)
{
// Signal the main thread to continue.
allDone.Set();
// Get the socket that handles the client request.
Socket listener = (Socket)ar.AsyncState;
Socket handler = listener.EndAccept(ar);
// Create the state object.
TcpStateObject state = new TcpStateObject();
state.workSocket = handler;
handler.BeginReceive(state.buffer, 0, TcpStateObject.BufferSize, 0,
new AsyncCallback(ReadCallback), state);
}
private void ReadCallback(IAsyncResult ar)
{
string content = string.Empty;
// Retrieve the state object and the handler socket
// from the asynchronous state object.
TcpStateObject state = (TcpStateObject)ar.AsyncState;
Socket handler = state.workSocket;
// Read data from the client socket.
int bytesRead = handler.EndReceive(ar);
if (bytesRead > 0)
{
// There might be more data, so store the data received so far.
state.sb.Append(Encoding.ASCII.GetString(
state.buffer, 0, bytesRead));
// Check for end-of-file tag. If it is not there, read
// more data.
content = state.sb.ToString();
int pos = content.IndexOf("<EOF>");
if (pos > -1)
{
// All the data has been read from the
// client. Place message into Queue for
// Unity to retrieve.
NetworkMessage message = new NetworkMessage();
message.content = content.Remove(pos);
message.clientAddress = ((IPEndPoint)handler.RemoteEndPoint).Address;
message.clientPort = ((IPEndPoint)handler.RemoteEndPoint).Port;
messageQueue.Enqueue(message);
}
else
{
// Not all data received. Get more.
handler.BeginReceive(state.buffer, 0, TcpStateObject.BufferSize, 0,
new AsyncCallback(ReadCallback), state);
}
}
}
private void PrintError(string e)
{
errorQueue.Enqueue(e);
}
// For container to check for networking errors.
public bool GetError(out string outError)
{
// Out promises initialization
return errorQueue.TryDequeue(out outError);
}
}
UDP Sending
public class UdpThreadedClient
{
// For returning error messages to main Unity thread
private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();
public void Send(string address, int port, string message)
{
var sendTask = new Task(() => SendTask(address, port, message));
sendTask.Start();
}
// Even though UDP sends immediately, taking very little time,
// it is technically blocking. There are very rare situations
// in which the call will take longer than usual, so sending
// takes place on a separate thread.
public void SendTask(string address, int port, string message)
{
using (UdpClient client = new UdpClient())
{
try
{
// String address to IPAddress
IPAddress ip;
IPAddress.TryParse(address, out ip);
// Convert message to byte array
byte[] buffer = Encoding.ASCII.GetBytes(message);
// Creatte IPEndPoint from the given address and port
IPEndPoint endPoint = new IPEndPoint(ip, port);
// Send the contents of buffer
client.Send(buffer, buffer.Length, endPoint);
}
catch (SocketException se)
{
PrintError("SocketException sending to " + address + ":" + port + ": " + se.ToString());
}
catch (Exception e)
{
PrintError("Unexpected exception sending to " + address + ":" + port + ": " + e.ToString());
}
}
}
private void PrintError(string e)
{
errorQueue.Enqueue(e);
}
// For container to check for networking errors.
public bool GetError(out string outError)
{
// Out promises initialization
return errorQueue.TryDequeue(out outError);
}
}
UDP Receiving
public class UdpThreadedServer
{
// Queue for sharing messages with Unity thread.
private ConcurrentQueue<NetworkMessage> messageQueue = new ConcurrentQueue<NetworkMessage>();
// For returning error messages to main Unity thread
private ConcurrentQueue<string> errorQueue = new ConcurrentQueue<string>();
// For reading new messages from Queue
// without exposing Queue to other classes.
public bool GetMessage(out NetworkMessage message)
{
if (messageQueue.TryDequeue(out message))
{
return true;
}
return false;
}
// Start the network listening thread.
public void StartListening(int port)
{
// Uses Task, but with LongRunning option it will create
// new thread, not use one from ThreadPool.
var listenTask = new Task(() => ListenTask(port), TaskCreationOptions.LongRunning);
listenTask.Start();
}
private void ListenTask(int port)
{
// Bind the socket to the local endpoint and listen for incoming connections.
try
{
UdpClient udpClient = new UdpClient(port);
IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Any, 0);
while (true)
{
Byte[] buffer = udpClient.Receive(ref remoteEndPoint);
string content = Encoding.ASCII.GetString(buffer);
NetworkMessage message = new NetworkMessage();
message.content = content;
message.clientAddress = remoteEndPoint.Address;
message.clientPort = remoteEndPoint.Port;
messageQueue.Enqueue(message);
}
}
catch (Exception e)
{
PrintError("Error opening UDP socket: " + e.ToString());
}
}
private void PrintError(string e)
{
errorQueue.Enqueue(e);
}
// For container to check for networking errors.
public bool GetError(out string outError)
{
// Out promises initialization
return errorQueue.TryDequeue(out outError);
}
}