Esponendo ConcurrentQueue come IObservable ?

Question

Mi chiedevo se è possibile utilizzare una coda (specificamente come ConcurrentQueue) come la fonte di un IObservable? Qualcosa di simile;

  Queue = new ConcurrentQueue<IMessage>();
  var xs = Queue.AsEnumerable().ToObservable();

  xs.Subscribe((IMessage msg) =>
     {
        Console.WriteLine("Msg :" + msg.subject);
     });

Credo che in realtà non ha senso, perché nulla è stato rimosse dalla coda. Sto cercando di implementare un processo di non-blocking che possono iscriversi a "messaggi" spinti agli osservatori, quindi l'uso di una coda. Sono sicuro che dovrei essere in grado di fare questo con RX, ma non riesco a ottenere la mia testa intorno ad esso!

Sarei interessato a qualche suggerimento su come questo potrebbe essere implementato. Grazie!

Answer 1

Hai ragione, la conversione di una coda (concomitante o semplice, non importa) sarebbe solo enumerare, ma non de-coda. "Real" implementazione è possibile, ma più complesso - vedere il link a una domanda simile ho chiesto sul forum RX (che è ancora una migliore fonte di informazioni su RX confronto con StackOverflow):

Come implementare una coda solo lavoratore produttore consumatore utilizzando RX?