Rx.Subject

Represents an object that is both an observable sequence as well as an observer. Each notification is broadcasted to all subscribed observers.

This class inherits both from the Rx.Observable and Rx.Observer classes.

Usage ##

The follow example shows the basic usage of an Rx.Subject.

var subject = new Rx.Subject();

var subscription = subject.subscribe(
    function (x) {
        console.log('Next: ' + x.toString());
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

subject.onNext(42);

// => Next: 42

subject.onNext(56);

// => Next: 56

subject.onCompleted();

// => Completed

Location

• rx.js

Subject Constructor ##

• constructor

Subject Class Methods ##

• create

Subject Instance Methods ##

• dispose

• hasObservers

Inherited Classes ##

• Rx.Observable

• Rx.Observer

Subject Constructor ##

Rx.Subject()

# Ⓢ

Creates a subject.

Example

var subject = new Rx.Subject();

var subscription = subject.subscribe(
    function (x) {
        console.log('Next: ' + x);
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

subject.onNext(42);
// => Next: 42

subject.onCompleted();
// => Completed

Location

• rx.js

Subject Class Methods ##

Rx.Subject.create(observer, observable)

# Ⓢ

Creates a subject from the specified observer and observable.

Arguments

1. observer (Observer): The observer used to send messages to the subject.

2. observable (Observable): The observable used to subscribe to messages sent from the subject.

Returns

(Subject): Subject implemented using the given observer and observable.

Example

/* Using a Web Worker to send and receive data via an Rx.Subject */

/* worker.js */

self.onmessage = function(e) {
    self.postMessage(e.data);
};

/* client.js */
var worker = new Worker('worker.js');

// Create observer to handle sending messages
var observer = Rx.Observer.create(
    function (data) {
        worker.postMessage(data);
    });

// Create observable to handle the messages
var observable = Rx.Observable.create(function (obs) {

    worker.onmessage = function (data) {
        obs.onNext(data);
    };

    worker.onerror = function (err) {
        obs.onError(err);
    };

    return function () {
        worker.close();
    };
});

var subject = Rx.Subject.create(observer, observable);

var subscription = subject.subscribe(
    function (x) {
        console.log('Next: ' + x);
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

subject.onNext(42);
// => Next: 42

Location

• rx.js

Subject Instance Methods ##

Rx.Subject.prototype.dispose()

# Ⓢ

Unsubscribe all observers and release resources.

Example

var subject = new Rx.Subject();

var subscription = subject.subscribe(
    function (x) {
        console.log('Next: ' + x.toString());
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

subject.onNext(42);
// => Next: 42

subject.onCompleted();
// => Completed

subject.dispose();

try {
    subject.onNext(56);
} catch (e) {
    console.log(e.message);
}

// => Object has been disposed

Location

• rx.js

Rx.Subject.prototype.hasObservers()

# Ⓢ

Indicates whether the subject has observers subscribed to it.

Returns

(Boolean): Returns true if the Subject has observers, else false.

Example

var subject = new Rx.Subject();

console.log(subject.hasObservers());
// => false

var subscription = subject.subscribe(
    function (x) {
        console.log('Next: ' + x.toString());
    },
    function (err) {
        console.log('Error: ' + err);
    },
    function () {
        console.log('Completed');
    });

console.log(subject.hasObservers());
// => true

Location

• rx.js