src/datapipe/DataPipe.js
import {EventEmitter2} from 'eventemitter2';
import {randomHash} from '../utils';
import {DataPipeStatus} from './DataPipeStatus';
/**
* Private fields addresses. These values are initialized when this file is loaded and hides
* private vars addressing outside this scope.
*/
var priv = {
id: Symbol('id'),
internalId: Symbol('internalId'),
type: Symbol('type'),
stack: Symbol('stack'),
participants: Symbol('participants'),
label: Symbol('label'),
status: Symbol('status'),
};
var getStatusIndex = function(status) {
switch (status) {
case DataPipeStatus.UNCONNECTED:
return 0;
case DataPipeStatus.CONNECTING:
return 1;
case DataPipeStatus.CONNECTED:
return 2;
case DataPipeStatus.REJECTED:
return 3;
case DataPipeStatus.DISCONNECTED:
return 4;
default:
}
};
/**
* Provides access to methods for managing an outgoing or an incoming data pipe
* DataPipe objects are obtained by calling the {Session#createDataPipe} method or
* handling the incomingDataPipe event of a connected {Session} instance.
*
* DataPipe allows data interchange between N users when pipe reaches 'connected' status.
*
* A DataPipe is identified by a label: shared string that will be available to any
* peer.
*
* This class emits next events:
* - 'data'
* - 'data-with-info'
* - 'status'
*
* 'data' event:
* - Received data from another pipe peer.
* - Payload is a string that represents received data.
*
* 'data-with-info' event:
* - Received data with aditional info from another peer.
* - Event payload has next structure: {data: <String>, from: <sender>, ts: <timestamp>}
*
* 'status' event:
* - Pipe status changed.
* - Payload is a DataPipe.STATUS element
*
* @interface
*
* @example How to create a DataPipe
* pipe = session.createDataPipe(to)
* pipe.on("data", function(dt) {
* console.log("received data", dt);
* });
* pipe.when(DataPipeStatus.DISCONNECTED).then(function(){...});
* pipe.connect().then(function() {
* // Connecting code
* console.log("pipe connected, ready to send/receive");
* pipe.send("Hello world")
* });
*/
export class DataPipe extends EventEmitter2 {
/**
* @private
*
* @throws {TypeError}
*
* @param {String} id pipe identifier. Optional. Default is a random value.
* @param {DataPipe.TYPE} type pipe type.
* @param {WacStack} stack stack instance.
* @param {Array<String>} participants pipe participants.
* @param {String} label pipe label. Optional. Default is 'generic'
*/
constructor(id = randomHash(), type, stack, participants, label = 'generic') {
super();
/* Syntax check */
if (!(participants instanceof Array)) {
throw new TypeError('invalid value for participants. An Array expected but an ' + typeof participants + ' received.');
}
/* Store values */
this[priv.id] = id;
this[priv.internalId] = null;
this[priv.type] = type;
this[priv.stack] = stack;
this[priv.participants] = participants;
this[priv.label] = label;
this[priv.status] = DataPipeStatus.UNCONNECTED;
}
/**
* @private
*/
set status(status) {
if (this[priv.status] != status) {
this[priv.status] = status;
this.emit('status', status);
}
}
/**
* Sets a private identifier for pipe implementation data.
* @private
*/
set internalId(value) {
this[priv.internalId] = value;
}
/**
* Gets a unique identifier for the pipe
*
* @type {String}
*/
get id() {
return this[priv.id];
}
/**
* Returns a private identifier for pipe implementation data
* @private
* @type {String}
*/
get internalId() {
if (this[priv.internalId] !== null) {
return this[priv.internalId];
} else {
return this.id;
}
}
/**
* Gets pipe type
*
* @type {DataPipe.TYPE}
*/
get type() {
return this[priv.type];
}
/**
* Gets stack
*
* @private
*
* @type {WacStack}
*/
get stack() {
return this[priv.stack];
}
/**
* Gets the identities of the peers attached to this pipe
*
* @type {Array<String>}
*/
get participants() {
return this[priv.participants];
}
/**
* Gets the identities of the remote peers attached to this pipe
*
* @type {Array<String>}
*/
get remoteParticipants() {
return this[priv.participants].filter((participant) => {
return participant !== this.stack.wStack.getCurrentCredential().data.username;
});
}
/**
* Return label associated with this DataPipe. Requires DataPipe to be connected
*
* @type {String}
*/
get label() {
return this[priv.label];
}
/**
* Retrieves the current status of this pipe.
*
* @type {DataPipeStatus}
*/
get status() {
return this[priv.status];
}
/**
* Adds a participant to this pipe.
*
* When this method fails returned {Promise} is rejected with one of next {Error}
* - 'unimplemented'
* - 'not allowed'
* - 'invalid-state' error when called on "disconnected" state
*
* @param {String} identity participant to add
*
* @returns {Promise<DataPipe, Error>} Resolved with the DataPipe object when
* invitation sent to new participant or rejected with an {Error}.
*/
addParticipant() {
return Promise.reject(new Error('unimplemented'));
}
/**
* Removes a participant from this pipe.
*
* When this method fails returned {Promise} is rejected with one of next {Error}
* - 'unimplemented'
* - 'not allowed'
* - 'invalid-state' error when called on "disconnected" state
*
* @param {String} identity participant to remove
*
* @returns {Promise<DataPipe, Error>} Resolved with the DataPipe object when
* participant was removed or rejected with an {Error}.
*/
removeParticipant() {
return Promise.reject(new Error('unimplemented'));
}
/**
* Attempts to reach the pipe recipient and establish a connection.
* For an incoming pipe, calling this method explicitly joins/accepts the pipe.
*
* When this method fails returned {Promise} is rejected with one of next {Error}
* - 'unimplemented'
* - 'invalid-state' error when called on state different than "unconnected" state
* - 'disconnected' pipe reaches disconnected state without being connected
*
* @return {Promise<DataPipe, Error>} Resolved with the DataPipe object when
* pipe reaches "connecting" state or rejected with an {Error}.
*/
connect() {
return Promise.reject(new Error('unimplemented'));
}
/**
* Ends an active pipe.
*
* When this method fails returned {Promise} is rejected with one of next {Error}
* - 'unimplemented'
* - 'invalid-state' error when called on state different than "connected" state
*
* @return {Promise<DataPipe, Error>} Resolved with the DataPipe object when
* pipe reaches "disconnected" state or rejected with an {Error}.
*/
disconnect() {
return Promise.reject(new Error('unimplemented'));
}
/**
* Called when a user does not wish to accept an incoming pipe.
*
* When this method fails returned {Promise} is rejected with one of next {Error}
* - 'unimplemented'
* - 'invalid-state' error when called on state different than "unconnected" state
* - 'invalid-state' error when called on outgoing pipes
*
* @return {Promise<DataPipe, Error>} Resolved with the DataPipe object when
* pipe reaches "disconnected" state or rejected with an {Error}.
*/
reject() {
return Promise.reject(new Error('unimplemented'));
}
/**
* Sends data to the DataPipe recipients
*
* When this method fails returned {Promise} is rejected with one of next {Error}
* - 'unimplemented'
* - 'invalid-state' error when called on state different than "connected" state
*
* @param {String} data data to send
*
* @return {Promise<DataPipe, Error>} Resolved with DataPipe object after data
* delivery or rejected with an {Error}
*/
send() {
return Promise.reject(new Error('unimplemented'));
}
/**
* Returns a Promise resolved when pipe reaches provided state
*
* - "unreachable state" pipe can not reach this state
* - "invalid state" pipe state changed to an state that makes impossible reach
* desired state.
*
* @param {DataPipeStatus} status desired state
*
* @return {Promise<DataPipe, Error>} Resolved with DataPipe object when pipe reaches
* provided state or rejected with an {Error}.
*/
when(status) {
let index = getStatusIndex(status);
return new Promise((resolve, reject) => {
if (getStatusIndex(this[priv.status]) > index) {
reject(new Error('unreachable state'));
} else if (getStatusIndex(this[priv.status]) == index) {
resolve(this);
} else {
var onStatus = (newStatus) => {
let newIndex = getStatusIndex(newStatus);
if (newIndex == index) {
this.off('status', onStatus);
resolve(this);
} else if (newIndex > index) {
this.off('status', onStatus);
reject(new Error('invalid state'));
}
};
this.on('status', onStatus);
}
});
}
}
