Is this an ugly/incorrect usage of subscriptions?

Little backstory: I'm trying to write code that will read messages from INetworkConnection that the INetworkConnection publishes and then send those up to the NetworkController which should do some processing and then send that off to an injected IObserver (which is a single observer because there's a global listener interested in network messages).

In other words, I'm trying to read messages, transform them, and then push them off to some listener. The reason why the controller is not subscribable is because there is only meant to be one listener (which will then distribute messages as necessary).

Now, right now I haven't added the transformation, so all the network controller does is that it pipes the messages from every connection into the injected subscriber like so:

private readonly IDisposable _networkConnectionStreamSubscription;
    private readonly IObserver _networkMessageStream;

    public NetworkController(IObservable networkConnectionStream,
        IObserver networkMessageStream)
    {
        _networkConnectionStreamSubscription = networkConnectionStream.Subscribe(
            x => LinkWith(x),
            () => Close());
        _networkMessageStream = networkMessageStream;
    }

    /// 
    /// Link the Controller with the given Connection, allowing the network
    /// controller to start listening for network messages.
    /// 
    /// 
    public void LinkWith(INetworkConnection networkConnection)
    {
        // Subscribe to the network message, and pipe the messages that have been parsed
        // out through our network message stream.
        // TODO: Do something with the subscription disposable
        networkConnection.Subscribe(
            x => _networkMessageStream.OnNext(x),
            exception => _NetworkMessageError(networkConnection, exception),
            () => _NetworkConnectionClosed(networkConnection)
        );
    }

My question is, simply speaking, am I

• going the wrong way by thinking of Rx as str

I don't think your approach is fundamentally wrong: Rx is all about exposing and responding to streams of data. I do wonder though if any of this is necessary; doesn't the Observable.Select method do essentially what you are trying to do?

StackExchange Code Review Q#51504, answer score: 2