CODEWITHSUNDEEP
c#.netasynchronousstream
Alex Shkor
Alex Shkor

Reputation: 1209

How to copy stream to many streams async C#.NET

I have TCP server , what recieve big data without a break. And I need to broadcast this stream to many clients.

UPDATE: I need to broadcast video stream. Perhaps there are ready solutions?

Upvotes: 3

Views: 1570

Answers (3)

casperOne
casperOne

Reputation: 74560

If you want to do this asynchronously, then you can take advantage of the System.Threading.Tasks namespace.

First, you'll need a map of the Stream instances to the Task that can be waited on for completion:

IDictionary<Stream, Task> streamToTaskMap = outputStreams.
    ToDictionary(s => s, Task.Factory.StartNew(() => { });

There's a slight overhead in the above, in that there's a wasted Task instance that does nothing, but that price is small given the number of Task instances and continuations you need to perform.

From there, you would read the content from the stream and then write it out to each of the Stream instances asynchronously:

byte[] buffer = new byte[<buffer size>];
int read = 0;

while ((read = inputStream.Read(buffer, 0, buffer.Length)) > 0)
{
    // The buffer to copy into.
    byte[] copy = new byte[read];

    // Perform the copy.
    Array.Copy(buffer, copy, read);

    // Cycle through the map, and replace the task with a continuation
    // on the task.
    foreach (Stream stream in streamToTaskMap.Keys)
    {
        // Continue.
        streaToTaskMap[stream] = streaToTaskMap[stream].ContinueWith(t => {
            // Write the bytes from the copy.
            stream.Write(copy, 0, copy.Length);
        });
    }
}

And in the end, you can wait on all of the streams written to by calling:

Task.WaitAll(streamToTaskMap.Values.ToArray());

There are a few things to note.

First, the copy of buffer is needed because of the lambda that is passed to ContinueWith; the lambda is a closure that would encapsulate buffer and because it is processing asynchronously, the contents are likely to change. Each continuance needs its own copy of the buffer to read.

This is also why the call to Stream.Write uses the Array.Length property; otherwise, the read variable would have to be copied through each iteration of the loop.

Additionally, it would be more ideal to be able to utilize the BeginWrite/EndWrite methods on the Stream class; because there is no ContinueWithAsync method which will take a Task and continue with an asynchronous method, there is no benefit to calling the async versions of read.

This is one of those cases where it might be better to call BeginWrite/EndWrite yourself (as well as BeginRead/EndRead) in order to make the most of async operations; of course, this will be a bit more complex because you won't have the encapsulation of the operation's result that Task provides, and you will have to take the same precautions with the buffer if you use anonymous methods/closures.

Upvotes: 2

Mranz
Mranz

Reputation: 1278

Spawn off a thread passing the stream to it, as well as the streams to write to

byte[] buffer = new byte[BUFFER_SIZE];
int btsRead = 0;

while ((btsRead = inputStream.Read(buffer, 0, BUFFER_SIZE)) > 0)
{
    foreach (Stream oStream in outputStreams)
        oStream.Write(buffer, 0, btsRead);
}

EDIT: To parallelize the writes:

replace the foreach chunk with:

Parallel.ForEach(outputStreams, oStream =>
{
    oStream.Write(buffer, 0, btsRead);
});

Upvotes: 1

uadrive
uadrive

Reputation: 1269

Basically, you are wanting to thread your application. Here's a simple example of threading and TCP/IP

C# Tutorial - Simple Threaded TCP Server | Switch on the Code

Upvotes: 0

Related Questions