Наблюдаемый TcpListener заканчивается после однократного соединения

Question

Я новичок в Rx, поэтому я, вероятно, делаю некоторые важные ошибки здесь.

Я хотел создать очень простой сервер сокетов, который мог бы получать сообщения от клиентов с помощью Observables. Для этого я использую Rxx, который предоставляет методы расширения в пространстве имен System.Net.Sockets, а также предоставляет статический заводский класс ObserableTcpListener.

Вот что я до сих пор в значительной степени украл ее из различных источников:

IPEndPoint endpoint = new IPEndPoint(IPAddress.Parse("127.0.0.1"), 9001); 
TcpListener listener = new TcpListener(endpoint); 

IObservable<TcpClient> clients = listener 
    .StartSocketObservable(1) 
    .SelectMany<Socket, TcpClient>(socket => SocketToTcpClient(socket)); 
    .Finally(listener.Stop) 

clients.Subscribe(client => 
{ 
    OnConnect(client).Subscribe(
     message => OnMessage(client, message), 
     ex => OnException(client, ex), 
     () => OnCompleted(client)); 
}); 

private static IObservable<TcpClient> SocketToTcpClient(Socket socket) 
{ 
    TcpClient client = new TcpClient(); 
    client.Client = socket; 
    return Observable.Return<TcpClient>(client); 
} 

private static IObservable<byte[]> OnConnect(TcpClient client) 
{ 
    return client.Client.ReceiveUntilCompleted(SocketFlags.None); 
} 

private static void OnMessage(TcpClient client, byte[] message) 
{ 
    Console.WriteLine("Mesage Received! - {0}", Encoding.UTF8.GetString(message)); 
} 

private static void OnCompleted(TcpClient client) 
{ 
    Console.WriteLine("Completed."); 
} 

private static void OnException(TcpClient client, Exception ex) 
{ 
    Console.WriteLine("Exception: {0}", ex.ToString()); 
} 

Это работает ... до точки. Я могу создать одно клиентское соединение. Как только это соединение завершается, кажется, что последовательность Observable завершается и вызывается .Finally(listener.Stop). Очевидно, это не то, что я хочу.

Я пробовал использовать фабричный класс ObserableTcpListener.Start(), но это дает мне тот же результат.

IObservable<TcpClient> sockets = ObservableTcpListener.Start(endpoint); 
sockets.Subscribe(client => 
{ 
    OnConnect(client).Subscribe(
     message => OnMessage(client, message), 
     ex => OnException(client, ex), 
     () => OnCompleted(client)); 
}); 

Я предполагаю, что я понимаю, проблема здесь: clients наблюдаемая последовательность просто пустой после того, как первый клиент прекращает, таким образом .Finally(listener.Stop) называется.

Что мне нужно сделать, чтобы обойти это? Как я могу продолжать слушать входящие соединения?

Answer 1

Сделайте Observableгорячим и сохраняйтесь, пока есть подписки.

IObservable<TcpClient> clients = listener 
    .StartSocketObservable(1) 
    .SelectMany<Socket, TcpClient>(socket => SocketToTcpClient(socket)) 
    .Finally(listener.Stop) 
    .Publish().RefCount();