4
\$\begingroup\$

I am creating a server client app where after the connection is done, the server and client will send packages back and forward. The Stream can be a NetworkStream or SslStream.

I have created a Async ReadContinuously method and it seems to work, but I do not trust my own knowledge about Async yet. Can you guys tell me I am on the right track or not?

Client:

        private async void ListenToServer()
        {
            bool exitbyerror = false;

            //
            Queue<TestServerDataPacket> queue = new Queue<TestServerDataPacket>();

            try
            {
                await Task.Run(() =>
                {
                    // After this a Queue Reader must be created
                    _packetReader.ReadContinuously(_netStream, _connection.ReceiveBufferSize, queue);

                    // For Testing
                    while (_connection?.Connected == true)
                    {
                        //
                        Console.WriteLine("Client: ({0}) Packets in queue.", queue.Count);

                        // For Testing 
                        Thread.Sleep(2000);
                    }
                });
            }
            catch
            {
                exitbyerror = true;
            }

            //
            if (exitbyerror)
            {
                //
            }
        }

PacketReader:

        private bool _readContinuously;

        public void ReadContinuously(Stream s, int bufferSize, Queue<TestServerDataPacket> packetQueue)
        {
            try
            {
                if (s == null)
                {
                    throw new ArgumentNullException("Stream can not be null!");
                }
                if (packetQueue == null)
                {
                    throw new ArgumentNullException("Queue<TestServerDataPacket> can not be null!");
                }

                //
                _readContinuously = true;

                //
                DoReadContinuously(s, bufferSize, packetQueue);
            }
            catch
            {
                throw;
            }
        }
        private async void DoReadContinuously(Stream s, int bufferSize, Queue<TestServerDataPacket> packetQueue)
        {
            //
            byte[] buffer = new byte[bufferSize];

            //
            TestServerDataPacket packet;

            try
            {
                // Read Packet Length = 4 bytes
                int bytesReceived = 0;
                while (bytesReceived < 4)
                {
                    //
                    int byteread = await s.ReadAsync(buffer, bytesReceived, 4 - bytesReceived);

                    //
                    if (byteread == 0)
                    {
                        // 0 bytes read = end of stream / disconnected
                        throw new Exception("Connection Closed!");
                    }

                    //
                    bytesReceived += byteread;
                }
                bytesReceived = 0;

                // Get Packet Size
                int packetSize = BitConverter.ToInt32(buffer, 0);

                // Create Packet Byte Array
                byte[] packetbytes;

                // Read Data
                using (MemoryStream memoryStream = new MemoryStream())
                {
                    // Read Data
                    while (bytesReceived < packetSize)
                    {
                        // Adjust Buffer size to catch only the packet and nothing else
                        if (buffer.Length > (packetSize - bytesReceived))
                        {
                            buffer = new byte[(packetSize - bytesReceived)];
                        }

                        //
                        int count;
                        if ((count = await s.ReadAsync(buffer, 0, buffer.Length)) > 0)
                        {
                            // Save Data
                            memoryStream.Write(buffer, 0, buffer.Length);

                            // Count
                            bytesReceived += count;
                        }
                    }

                    // Get Packet Bytes Array
                    packetbytes = memoryStream.GetBuffer();
                }

                // Create Packet
                DeserializeData(packetbytes, out packet);
            }
            catch
            {
                throw;
            }

            //
            if (packet != null)
            {
                packetQueue.Enqueue(packet);
            }

            //
            if (_readContinuously)
            {
                DoReadContinuously(s, bufferSize, packetQueue);
            }
        }

Working test server:

namespace TestServer
{
   public class Program
    {
        static void Main(string[] args)
        {
            TestServer server = new TestServer();

            TestClient client = new TestClient();

            Console.ReadLine();
        }
    }

    public class TestServer
    {
        private readonly TcpListener _listener;

        public TestServer()
        {
            IPEndPoint localEndPoint = new IPEndPoint(IPAddress.Any, 45654);

            _listener = new TcpListener(localEndPoint);
            _listener.Start(100);

            AcceptConnections();
        }
        private async void AcceptConnections()
        {
            await Task.Run(async () =>
            {
                try
                {
                    Socket s = await _listener.AcceptSocketAsync();
                    if (s != null)
                    {
                        Console.WriteLine("Server: Client Connected");
                        TestServerConnection c = new TestServerConnection(s);
                    }
                }
                catch
                {
                    //
                }
            });
        }
    }

    public class TestServerConnection
    {
        private readonly Socket _connection;
        private readonly TestPacketSender _packetSender;

        public TestServerConnection(Socket s)
        {
            _connection = s;

            _packetSender = new TestPacketSender(new NetworkStream(s, FileAccess.ReadWrite));

            Task.Factory.StartNew(ListenToClient);
        }
        private async Task ListenToClient()
        {
            bool exitbyerror = false;

            try
            {
                await Task.Run(() =>
                {
                    while (_connection?.Connected == true)
                    {
                        Thread.Sleep(10000);

                        Console.WriteLine("Server: Sending Hi");
                        _packetSender.Send(new TestServerDataPacket(2000)); // Int 2000 is Hi
                    }
                });
            }
            catch
            {
                exitbyerror = true;
            }

            //
            if (exitbyerror)
            {
                //
            }
        }
    }

    public class TestClient
    {
        private readonly Socket _connection;
        private NetworkStream _netStream;
        private TestPacketReader _packetReader;

        public TestClient()
        {
            _packetReader = new TestPacketReader();

            IPEndPoint remoteEndPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 45654);

            // Create a TCP/IP socket.  
            _connection = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)
            {
                ReceiveBufferSize = (8 * 1024),
                SendBufferSize = (8 * 1024),
                NoDelay = true
            };

            // Connect to the remote endpoint.  
            _connection.Connect(remoteEndPoint);
            _netStream = new NetworkStream(_connection, FileAccess.ReadWrite);

            Console.WriteLine("Client: Connected to server.");

            Task.Factory.StartNew(ListenToServer);
        }
        private async void ListenToServer()
        {
            bool exitbyerror = false;

            //
            Queue<TestServerDataPacket> queue = new Queue<TestServerDataPacket>();

            try
            {
                await Task.Run(() =>
                {
                    // After this a Queue Reader must be created
                    _packetReader.ReadContinuously(_netStream, _connection.ReceiveBufferSize, queue);

                    // For Testing
                    while (_connection?.Connected == true)
                    {
                        //
                        Console.WriteLine("Client: ({0}) Packets in queue.", queue.Count);

                        // For Testing 
                        Thread.Sleep(2000);
                    }
                });
            }
            catch
            {
                exitbyerror = true;
            }

            //
            if (exitbyerror)
            {
                //
            }
        }
    }

    public class TestPacketSender
    {
        //
        private readonly Stream _stream;
        private readonly object _writingToStream = new object();

        //
        public TestPacketSender(Stream stream)
        {
            _stream = stream;
        }

        //
        public bool Send(TestServerDataPacket packet)
        {
            if (_stream == null)
            {
                throw new ArgumentNullException("Stream can not be null");
            }
            if (packet == null)
            {
                throw new ArgumentNullException("TestServerDataPacket can not be null!");
            }

            //
            lock (_writingToStream)
            {
                return SendToStream(_stream, packet);
            }
        }
        private bool SendToStream(Stream s, TestServerDataPacket packet)
        {
            try
            {
                // Byte Array containing Packet Size and Packet
                byte[] buffer;

                // Fill buffer (Packet Size + Packet Content)
                SerializeData(packet, out buffer);

                // Write Packet to the Stream
                s.Write(buffer, 0, buffer.Length);
                s.Flush();

                //
                return true;
            }
            catch
            {
                //
                return false;
            }
        }

        //
        private void SerializeData(TestServerDataPacket packet, out byte[] buffer)
        {
            if (packet == null)
            {
                buffer = new byte[0];
                return;
            }

            byte[] packetbytes;

            BinaryFormatter formatter = new BinaryFormatter();
            using (MemoryStream ms = new MemoryStream())
            {
                //
                formatter.Serialize(ms, packet);

                //
                packetbytes = ms.ToArray();
            }

            //
            buffer = CreatePacket(packetbytes);
        }
        private byte[] CreatePacket(byte[] packetbytes)
        {
            // Get the packet length
            byte[] lengthPrefix = BitConverter.GetBytes(packetbytes.Length);

            // 
            byte[] totalpacket = new byte[lengthPrefix.Length + packetbytes.Length];

            // Combine the packet length and the packet data
            lengthPrefix.CopyTo(totalpacket, 0);
            packetbytes.CopyTo(totalpacket, lengthPrefix.Length);

            //
            return totalpacket;
        }
    }
    public class TestPacketReader
    {
        private bool _readContinuously;

        public void ReadContinuously(Stream s, int bufferSize, Queue<TestServerDataPacket> packetQueue)
        {
            try
            {
                if (s == null)
                {
                    throw new ArgumentNullException("Stream can not be null!");
                }
                if (packetQueue == null)
                {
                    throw new ArgumentNullException("Queue<TestServerDataPacket> can not be null!");
                }

                //
                _readContinuously = true;

                //
                DoReadContinuously(s, bufferSize, packetQueue);
            }
            catch
            {
                throw;
            }
        }
        private async void DoReadContinuously(Stream s, int bufferSize, Queue<TestServerDataPacket> packetQueue)
        {
            //
            byte[] buffer = new byte[bufferSize];

            //
            TestServerDataPacket packet;

            try
            {
                // Read Packet Length = 4 bytes
                int bytesReceived = 0;
                while (bytesReceived < 4)
                {
                    //
                    int byteread = await s.ReadAsync(buffer, bytesReceived, 4 - bytesReceived);

                    //
                    if (byteread == 0)
                    {
                        // 0 bytes read = end of stream / disconnected
                        throw new Exception("Connection Closed!");
                    }

                    //
                    bytesReceived += byteread;
                }
                bytesReceived = 0;

                // Get Packet Size
                int packetSize = BitConverter.ToInt32(buffer, 0);

                // Create Packet Byte Array
                byte[] packetbytes;

                // Read Data
                using (MemoryStream memoryStream = new MemoryStream())
                {
                    // Read Data
                    while (bytesReceived < packetSize)
                    {
                        // Adjust Buffer size to catch only the packet and nothing else
                        if (buffer.Length > (packetSize - bytesReceived))
                        {
                            buffer = new byte[(packetSize - bytesReceived)];
                        }

                        //
                        int count;
                        if ((count = await s.ReadAsync(buffer, 0, buffer.Length)) > 0)
                        {
                            // Save Data
                            memoryStream.Write(buffer, 0, buffer.Length);

                            // Count
                            bytesReceived += count;
                        }
                    }

                    // Get Packet Bytes Array
                    packetbytes = memoryStream.GetBuffer();
                }

                // Create Packet
                DeserializeData(packetbytes, out packet);
            }
            catch
            {
                throw;
            }

            //
            if (packet != null)
            {
                packetQueue.Enqueue(packet);
            }

            //
            if (_readContinuously)
            {
                DoReadContinuously(s, bufferSize, packetQueue);
            }
        }

        private void DeserializeData(byte[] data, out TestServerDataPacket packet)
        {
            BinaryFormatter formatter = new BinaryFormatter();
            using (MemoryStream stream = new MemoryStream())
            {
                //
                stream.Write(data, 0, data.Length);
                stream.Seek(0, SeekOrigin.Begin);

                //
                packet = (TestServerDataPacket)formatter.Deserialize(stream);
            }
        }
    }

    [Serializable]
    public class TestServerDataPacket
    {
        // Unique Id
        public readonly Guid Id;

        // Type
        public readonly TestServerPacketType Type;

        // Sugnal/Message
        public readonly int Signal = 0;

        //
        public TestServerDataPacket(int signal)
        {
            Id = Guid.NewGuid();

            Signal = signal;
            Type = TestServerPacketType.Signal;
        }
    }

    public enum TestServerPacketType
    {
        Signal
    }
}
\$\endgroup\$
1
  • \$\begingroup\$ async void amounts to "fire and forget". I'm surprised to see so much of it. Are you sure you shouldn't be awaiting Tasks more often? \$\endgroup\$
    – RubberDuck
    Aug 22 '17 at 18:11
3
\$\begingroup\$

You are on the right way.

There should be a loop at accepting clients otherwise you will accept only one client until you restart the server.

And there is a issue at the PacketReader part when s.ReadAsync returns a part instead of the whole packet. Aswell no need to recreate the buffer to the total size.

Here the corrected part:

            // Read Data
            using (MemoryStream memoryStream = new MemoryStream())
            {
                int count;
                // Read Data
                while (bytesReceived < packetSize)
                {
                    if ((count = await s.ReadAsync(buffer, 0, packetSize - bytesReceived > buffer.Length ? buffer.Length : packetSize - bytesReceived)) > 0)
                    {
                        // Save Data
                        memoryStream.Write(buffer, 0, count);

                        // Count
                        bytesReceived += count;
                    }
                }

                // Get Packet Bytes Array
                packetbytes = memoryStream.GetBuffer();
            }
\$\endgroup\$
2
  • \$\begingroup\$ Thnx, I will update my code. \$\endgroup\$
    – Eric S.
    Sep 19 '17 at 10:42
  • \$\begingroup\$ In a way, doesn't putting the read command in a loop defeat the purpose of the asynchronous operation? If bytes keep coming in sporadically, you just keep circling the while a little like a busy-waiting loop until it fills (albeit, returning control to the caller somewhat). Ideally, ReadAsync could be instructed to only return from await with >= someNumberOfBytes, but I don't think this exists. I'm not just trying to criticize btw, I'm having a similar issue myself and looking for the best solution. Is there a better way? \$\endgroup\$
    – nivk
    Mar 12 '19 at 1:47
1
\$\begingroup\$

For your topic, you may look at these documents (note the language dropdown on the right-side): Asynchronous Server Socket Example and Asynchronous Client Socket Example.

Use var for obvious assignments

When creating variables in short-lived scopes (e.g. methods, loops, using, etc.), immediately assigning a value or reference, and the assigned type is obvious the keyword var makes code more writable and lessens the distractions of typing.

// Obvious variable declaration and assignment/initialization
var localEndPoint = new IPEndPoint(IPAddress.Any, 45654);
using (var memoryStream = new MemoryStream()) { // omitted }
for (var i = 0; i < anObject.Length; i++) { // omitted }

// Not obvious or uninitialized
var processor = this.manager.Processor; // requires more than a quick glance
var counter;  // compile error

Thus the main method becomes:

static void Main(string[] args)
{
    var server = new TestServer();

    var client = new TestClient();

    Console.ReadLine();
}

Private variable names

Don't prefix private class member variables with an underscore _. This is a relic of previous versions of C# and were a workaround to getting to class members in Intellisense. It is better (more readable) to simply follow camelCasingRules for private class members and to address them using the this keyword (or base depending on context). Using this will narrow the Intellisense (or other selector) and will clearly indicate which object's members you are referring (as well as let repetition of names in narrower scopes).

Typcial usage of this

public class AnotherServer
{
    private readonly int delayInMilliseconds;
    public AnotherServer(int delayInMilliseconds)
    {
        this.delayInMilliseconds = delayInMilliseconds;
    }
}

Maing the code (less writable) more clear:

public class TestServer
{
    private readonly TcpListener listener;
    public TestServer()
    {
        var localEndPoint = new IPEndPoint(IPAddress.Any, 45654);

        this.listener = new TcpListener(localEndPoint);
        this.listener.Start(100);

        this.AcceptConnections();
    }
}

Avoid magic numbers

Using literal values is usually always a bad idea. I understand the need to use them for testing, but abstracting them into variables documents what you are trying to do. Descriptive variable names for values gives context and will remind you (or another programmer) what the code is doing. Plus the obvious reason is when you change a value, you need to make sure it is changed everywhere.

int defaultPortNumber = 45654;
// elsewhere
var localEndPoint = new IPEndPoint(IPAddress.Any, defaultPortNumber);
var remoteEndPoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), defaultPortNumber);

Also seeing some of these values leads me to believe you aren't aware of documented defaults (Socket.SendBufferSize and Socket.ReceiveBufferSize default to 8192 = 8 * 1024)

// Create a TCP/IP socket.  
this.connection = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp)
{
     NoDelay = true
};

Make sure all IDisposable objects get disposed

Continuing with the Socket code snippet, Socket implements IDisposable. You've enclosed other managed resources in the syntactical sugar that is using, but there's no clean up for the sockets you are using. Be wary of memory leaks!

Don't simply throw in a catch

This deserves its own type of punishment, but never catch an exception only to throw it! There's no point. It causes so many headaches down the road.

\$\endgroup\$
1
  • \$\begingroup\$ Thank you for all the tips and explanations, I will use them in future code. \$\endgroup\$
    – Eric S.
    Sep 19 '17 at 10:43

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.