Multi-Threading, Silverlight Sockets & Visualisation Part 2

Having successfully loaded my Visualisation data in Part 1, this article will demonstrate how to send binary encoded data over a socket to a Silverlight application.

Silverlight 2 has supports for asynchronous sockets, which can list to a data-stream connected to servers on a port between 4502 and 4532.  See Dan Wahlin’s good article on setting up a server for sending text messages.

My sample required sending a large amount of data the Model points + triangles, as well as the raw image stream.  This proved to be a challenge with sockets, as there is no well defined beginning or end point of a message, and this volume of data doesn’t can result in an almost constant stream of information.

The solution was to define a magic byte that what identify the start of the ‘message’, which was immediately followed by the size of the Body (as XML meta-data) and the image as binary stream:

Magic(1)

BodySize

ImageSize

Body

Image

Magic(2)

A BinaryWriter was used to send this formatted data to all listening clients:

    private const UInt32 Magic = 0xCAB2FAC;

    private void SendData(string data, byte[] image)
    {
        if (_clientStreams != null)
        {
            lock (_clientStreams)
            {
                List<BinaryWriter> deadWriters = new List<BinaryWriter>();
                // Write the data
                foreach (BinaryWriter writer in _clientStreams)
                {
                    if (writer != null)
                    {
                        try
                        {
                            // Get the body
                            byte[] body = Encoding.UTF8.GetBytes(data);

                            // Get bytes together
                            List<byte> bytes = new List<byte>();
                            bytes.AddRange(BitConverter.GetBytes(Magic));
                            bytes.AddRange(BitConverter.GetBytes((UInt32)body.Length));
                            bytes.AddRange(BitConverter.GetBytes((UInt32)image.Length));
                            bytes.AddRange(body);
                            bytes.AddRange(image);

                            // Write the out
                            writer.Write(bytes.ToArray());
                        }
                        catch (Exception)
                        {
                            // TODO: Trap specific connection exception
                            deadWriters.Add(writer);
                        }
                    }
                }
                // Remove the dead writers
                foreach (BinaryWriter writer in deadWriters)
                {
                    writer.Close();
                    _clientStreams.Remove(writer);
                }
            }
        }
    }

Silver 2 provides an asynchronous socket implementation which once connected can be wired up to receive events when data is transferred:

    void Page_Loaded(object sender, RoutedEventArgs e)
    {
        try
        {
            string host = Application.Current.Host.Source.DnsSafeHost;
            int ip = 4530;

            DnsEndPoint endPoint = new DnsEndPoint(host, ip);
            Socket socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);

            reader = new VisualisationReader();
            reader.Complete += new EventHandler<VisualisationReadEventArgs>(reader_Complete);

            SocketAsyncEventArgs args = new SocketAsyncEventArgs();
            args.UserToken = socket;
            args.RemoteEndPoint = endPoint;
            args.Completed += new EventHandler<SocketAsyncEventArgs>(OnSocketConnectCompleted);
            socket.ConnectAsync(args);
        }
        catch (Exception ex)
        {
            System.Diagnostics.Debug.WriteLine("Load Exception " + ex);
        }
    }

    private void OnSocketConnectCompleted(object sender, SocketAsyncEventArgs e)
    {
        // Read initial data
        byte[] buffer = new byte[1024];
        e.SetBuffer(buffer, 0, buffer.Length);
        if (e.SocketError == SocketError.Success)
        {
            e.Completed -= new EventHandler<SocketAsyncEventArgs>(OnSocketConnectCompleted);
            e.Completed += new EventHandler<SocketAsyncEventArgs>(OnSocketReceive);
            ReadMoreData(e);
        }
    }

    private void OnSocketReceive(object sender, SocketAsyncEventArgs e)
    {
        try
        {
            if (e.BytesTransferred > 0)
            {
                reader.Read(e);
            }
            // Read More
            ReadMoreData(e);
        }
        catch (Exception ex)
        {
            System.Diagnostics.Debug.WriteLine("Exception " + ex);
        }
    }

    void ReadMoreData(SocketAsyncEventArgs e)
    {
        Socket socket = (Socket)e.UserToken;
        if (!socket.ReceiveAsync(e))
        {
            // Check this for stack overflow
            OnSocketReceive(socket, e);
        }
    }

    void reader_Complete(object sender, VisualisationReadEventArgs e)
    {
        try
        {
            // Get the data set
            TrainingSetData d = GetData(e.Body);
            // Invoke
            Action<TrainingSetData, byte&#91;&#93;> handler = RenderData;
            this.Dispatcher.BeginInvoke(handler, new object[] { d, e.Image });
        }
        catch (Exception ex)
        {
            System.Diagnostics.Debug.WriteLine("Error reading body: " + e.Body);
            System.Diagnostics.Debug.WriteLine(ex);
            throw;
        }
    }

My sample wraps up the logic of interpreting the binary stream into a visualisation reader which calls back to the client once a message has been received.

    /// <summary>
    /// Reads the socket data stream
    /// </summary>
    public class VisualisationReader
    {

        /// <summary>
        /// Token to track the progress of reading the socket data
        /// </summary>
        protected enum Token
        {
            None,
            Header,
            Body,
            Image,
            Complete
        }

        /// <summary>
        /// Header struct to contain the structure of the message
        /// </summary>
        protected struct Header
        {
            public Header(uint body, uint pixels)
            {
                BodyLength = (int)body;
                ImagePixels = (int)pixels;
            }
            public int BodyLength;
            public int ImagePixels;
        }

        public event EventHandler<VisualisationReadEventArgs> Complete;

        public void Read(SocketAsyncEventArgs e)
        {
            e.SetBuffer(buffer, 0, buffer.Length);
            transfered = e.BytesTransferred;

            System.Diagnostics.Debug.WriteLine("Reading... {0}", transfered);

            // Reset the number of bytes read
            read = 0;

            while (read < transfered)
            {
                Token token = NextToken();
                if (token == Token.Complete)
                {
                    OnComplete(new VisualisationReadEventArgs(body.ToString(), image));
                    ResetToken();
                    System.Diagnostics.Debug.WriteLine("Break...");
                    break; // Break out of this read
                }
                else if (token == Token.None)
                {
                    // Exit if we didn't match anything
                    break;
                }
            }

        }

        /// <summary>
        /// Raises the SomeEvent event
        /// </summary>
        protected virtual void OnComplete(VisualisationReadEventArgs e)
        {
            EventHandler<VisualisationReadEventArgs> handler;
            lock (this)
            {
                handler = Complete;
            }
            if (handler != null)
            {
                handler(this, e);
            }
        }

        protected Token NextToken()
        {
            if (currentToken == Token.None)
            {
                // Wait until i get a magic byte at the begining of the buffer
                if (BitConverter.ToUInt32(buffer, 0) == Magic)
                {
                    currentToken = Token.Header;
                    System.Diagnostics.Debug.WriteLine("Read Magic");
                    read = 4;
                }
            }
            else if (currentToken == Token.Header)
            {
                // Build the header for body + image size
                header = new Header(BitConverter.ToUInt32(buffer, 4), BitConverter.ToUInt32(buffer, 8));
                read += 8;

                // Initialize for body
                currentToken = Token.Body;
                readTo = read + header.BodyLength;
                body = new StringBuilder(header.BodyLength);

                System.Diagnostics.Debug.WriteLine("Read Header");
            }
            else
            {
                // ... Implementation Detail for Body and Image ...
            }
        }

        private void ResetToken()
        {
            currentToken = Token.None;
        }

    }

That complete call back contains all the necessary information for rendering the image and overlaying the data points on the sliverlight canvas.  Part 3 will cover this in more detail including an alternative implementation for downloading an image on demand over a RESTful WCF web service.

The source is avaialble for download.

---