Create Rx.NET Observable in SignalR hub

Question

I have a SignalR hub that listens to client requests and uses Rx.NET to observe a database table, in order to send back updates to the client that requested them, as soon as they are available. But it looks like the Observer instance that is created in the hub, according to the client request, is destroyed (by the GC?) as soon as the method call finishes; therefore, I get no updates.

This is my current Hub implementation:

public class BookHub : Hub {
    private readonly BookService _service = new BookService();

    public void RequestBookUpdate(string author) {
        BookObserver observer = new BookObserver(Context.connectionId);
        IDisposable unsubscriber = _service.RequestBookUpdate(author, observer);
    }
}

The BookService returns a LINQ query converted to an Observable:

public IDisposable RequestBookUpdate(string author, BookObserver observer) {
    var query = from b in db.Book where b.Author.Contains(author) select b;
    IObservable<Book> observable = query.ToObservable();
    IDisposable unsubscriber = observable.Subscribe(observer);
    return unsubscriber;
}

The BookObserver simply sends back the new retrieved items to the specific client who requested the updates (identified by the connectionId):

// omissis

private static readonly IHubContext _context = GlobalHost.ConnectionManager.GetHubContext<BookHub>();
private readonly string _connectionId;

public BookObserver(string connectionId) {
    connectionId = _connectionId:
}

public void OnNext(Book value) {
    _context.Clients.Client(_connectionId).foundNewBook(value);
}

I don't care about the BookService instance being destroyed, but I want the BookObserver to stay alive, so I can call unsubscriber.Dispose() only when the client disconnects. Is this possible?

Answer 1

The Observer is being automatically disposed when it receives a call to OnComplete. This is actually a very good pattern, as it means you don't have to manually dispose of Subscriptions like this:

Observable.Range(0, 100)
    .Subscribe(...);

or

Observable.Interval(TimeSpan.FromSeconds(1))
    .Take(10)
    .Subscribe();

So, to make sure your observer doesn't get disposed until it wants to be disposed, you can concat another empty, never completing, observable to your source.

IObservable<Book> observable = query.ToObservable()
    .Concat(Observable.Never<Book>());

However, depending on what you're attempting to do, it may be better to handle this elsewhere, such as the client.