Do I need to call dispose on directive $destroy event?

[erceth]

I have a situation where I have multiple directives that subscribe to a stream from a service. If a directive is destroyed do I need to unsubscribe from the stream so I don't have zombie listeners? Or does it get garbage collected. If I need to unsubscribe how do I do it? newZoneStream.dispose(); is undefined.

/*** Directive ***/
function panelController(rxService) {
    var vm = this;

    //private variables
    var newZoneStream = null;

    //public variables
    vm.zones = null;

    //public functions
    vm.init = init;
    vm.destroy = destroy;

    function init() {
        newZoneStream = rxService.getNewZoneStream();
        newZoneStream.subscribe(function(data) {
            vm.zones = data;
        });
    }

    function destroy() {
        // DO I NEED TO DISPOSE HERE?: newZoneStream.dispose(); ?
    }
}

function panelLink(scope, element, link, ctrl) {
    ctrl.init();
    scope.$on('$destroy', function() {
        ctrl.destroy();
    })

}


/*** Service ***/
function getNewZoneStream() {
    if (!newZonesStream) {
        newZonesStream = rx.Observable.create(function(observable) {
            socket.on("new-zones", function(data) {
                observable.onNext(data);
            });

            //clean up function
            return function() {
                //$sails.off("new-zones");
                console.log("new zones stream closed");
            }
        });
    }

    return newZonesStream;
}

[erceth]

User error. rx.Observable() returns a stream not subscription. You can't call .dispose() on a stream. I needed to call subscribe() on the stream to get a subscription object. I call .dispose() on the subscription. Here's what I replaced it with:

/*** Directive ***/
function init() {
            var newZoneStream = rxService.getNewZoneStream();
            newZoneSubscription = newZoneStream.subscribe(function (data) {
                console.log(vm.id);
                vm.zones = data;
                if (!$scope.$$phase) {
                    $scope.$apply();
                }
            });
        }
function destroy() {
            newZoneSubscription.dispose(); //stops listening to the stream
        }

/*** Service ***/
function getNewZoneStream () {
        if (!newZonesStream) {
            newZonesStream = Rx.Observable.create(function(observable) {
                socket.on("new-zones", function(data) {
                    observable.onNext(data);
                });

                //clean up function
                return function() {
                    //$sails.off("new-zones");
                    console.log("unsubscribed from stream");
                }
            });
        } 

        return newZonesStream;
    }