Observable.FromAsyncPattern mit UDPClient.EndReceive und ref remote Endpunkt Parameter

Question

Ich lerne über Reaktive Erweiterungen und versuche, etwas von meinem Code neu zu factorisieren.

UDPClient.EndReceive nimmt einen ref IPEndPoint Parameter, so habe ich zur Zeit diese Arbeit:

UdpClient receiverUDP = new UdpClient(); 
receiverUDP.Client.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.ReuseAddress, true); 
receiverUDP.EnableBroadcast = true; 
receiverUDP.Client.ExclusiveAddressUse = false; 
receiverUDP.Client.Bind(new IPEndPoint(IPAddress.Any, 1234)); 

IPEndPoint ep = null; 
var async = Observable.FromAsyncPattern<byte[]>(receiverUDP.BeginReceive, (i) => receiverUDP.EndReceive(i, ref ep)); 
var subscr = async().Subscribe(x => Console.WriteLine(ASCIIEncoding.ASCII.GetString(x))); 

Was passiert, wenn meine Abonnenten Zugriff auf die Remote IPEndPoint benötigen? In meiner aktuellen Inkarnation verwende ich Ereignisse und gebe eine benutzerdefinierte Klasse zurück, die byte[] und IPEndPoint umschließt. Ich kann nicht für das Leben von mir, erarbeiten, wie man das mit Rx macht.

Answer 1

Für jemand anderes suchen, ist es eine etwas einfachere und modernere Art und Weise zu tun dies mit ReceiveAsync:

public static IObservable<UdpReceiveResult> UdpStream(IPEndPoint endpoint) 
{ 
    return Observable.Using(() => new UdpClient(endpoint), 
     udpClient => Observable.Defer(() => 
      udpClient.ReceiveAsync().ToObservable()).Repeat()); 
} 

Sie können es mit IPAddress.Any nennen:

var stream = UdpStream(new IPEndPoint(IPAddress.Any, 514)); 

und verwenden Sie dann Select, um den Stream in den gewünschten Dateityp zu projizieren.

Answer 2

Wenn Sie bereits eine Wrapper-Klasse für byte[] und IPEndPoint erstellt warum nicht zurück, dass als Folge Select mit:

private IObservable<RemoteData> GetRemoteDataAsync() 
{ 
    return Observable.Defer(() => 
    { 
     UdpClient receiverUDP = new UdpClient(); 
     receiverUDP.Client.SetSocketOption(SocketOptionLevel.Socket, 
      SocketOptionName.ReuseAddress, true); 
     receiverUDP.EnableBroadcast = true; 
     receiverUDP.Client.ExclusiveAddressUse = false; 
     receiverUDP.Client.Bind(new IPEndPoint(IPAddress.Any, 1234)); 

     IPEndPoint ep = null; 
     return Observable.FromAsyncPattern<byte[]>(
        receiverUDP.BeginReceive, 
        (i) => receiverUDP.EndReceive(i, ref ep) 
       )() 
       .Select(bytes => new RemoteData(bytes, ep)); 
    }); 
}