Modify Rx event in C# before and after it is served to a subscriber

Viewed 278

I need to pre-process and post-process an event every time it gets observed. Here is the thing I came up with, that works:

var subj = new Subject<Event>();
var observable = Observable.Create<Event>(obs =>
{
    subj.Finally(obs.OnCompleted); 
    return subj.Subscribe(e =>
    {
        try
        {
            Preprocess(e);
            obs.OnNext(e);
            Postprocess(e);
        }
        catch (Exception ex) { obs.OnError(ex); }
    });
});

My question: is this the right way to do it, or maybe there's a better template/extension method for this?

1 Answers
Related