Observable.FromAsyncPattern avec UDPClient.EndReceive et le point de terminaison distant ref Paramètre

Question

Je suis en train d'étudier les extensions réactives et d'essayer de factoriser une partie de mon code.

UDPClient.EndReceive prend un paramètre ref IPEndPoint, j'ai donc actuellement ce travail:

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))); 

si mes abonnés ont besoin d'accéder à la IPEndPoint à distance? Dans mon incarnation actuelle, j'utilise des événements et je repasse une classe personnalisée qui enveloppe byte[] et IPEndPoint. Je ne peux pas pour la vie de moi, travailler sur comment faire cela avec Rx.

Answer 1

Pour ceux qui cherchent d'autre, il y a un moyen de faire un peu plus simple et plus moderne cela en utilisant ReceiveAsync:

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

Vous pouvez l'appeler avec IPAddress.Any:

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

puis utilisez Select pour projeter le flux vers le type de votre choix.

Answer 2

Si vous avez déjà créé une classe wrapper pour byte[] et IPEndPoint pourquoi ne pas revenir que la séquence en utilisant Select:

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)); 
    }); 
}