Orleans Udp Stream

UdpBootstrapProvider.cs

namespace OrleansUdpStream
{
    using System;
    using System.Net.Sockets;
    using System.Threading.Tasks;
    using Orleans;
    using Orleans.Providers;
    using Orleans.Streams;

    public class UdpBootstrapProvider : IBootstrapProvider
    {
        private UdpClient _listener;
        private IStreamProvider _streamProvider;

        public string Name { get; private set; }
        
        public Task Close()
        {
            _listener.Dispose();
            _listener = null;
            return TaskDone.Done;
        }

        public Task Init(string name, IProviderRuntime providerRuntime, IProviderConfiguration config)
        {
            Name = name;
            GrainClient.Initialize();
            _streamProvider = GrainClient.GetStreamProvider("Default");
            _listener = new UdpClient(config.GetIntProperty("Port", 0));
            Task.Run(Listen);
            return TaskDone.Done;
        }

        private async Task Listen()
        {
            while (_listener != null)
            {
                await HandleResult(await _listener.ReceiveAsync());
            }
        }

        private async Task HandleResult(UdpReceiveResult result)
        {
            var deviceGuid = GetDeviceGuidFromBuffer(result.Buffer);
            var stream = _streamProvider.GetStream<byte[]>(deviceGuid, "MyNamespace");
            await stream.OnNextAsync(result.Buffer);
        }

        private Guid GetDeviceGuidFromBuffer(byte[] buffer)
        {
            // TODO: Map bytes from buffer to guid.
            return Guid.Empty;
        }
    }
}