yonote/yjs
24
0
Fork 0
Code Pull Requests 1 Releases Activity

added some Operations, a connector, more structure. In particular I put a lot of time into the event handling

Browse Source
This commit is contained in:
Kevin Jahns 2015-06-21 14:55:39 +02:00
parent 042bcee482
commit 3142b0f161
8 changed files with 634 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

446
src/Connectors.js Normal file
View File

@ -0,0 +1,446 @@
(function(){
function WebRTC(webrtc_options){
if(webrtc_options === undefined){
throw new Error("webrtc_options must not be undefined!")
}
var room = webrtc_options.room;
// connect per default to our server
if(webrtc_options.url === undefined){
webrtc_options.url = "https://yatta.ninja:8888";
}
var swr = new SimpleWebRTC(webrtc_options);
this.swr = swr;
var self = this;
var channel;
swr.once('connectionReady',function(user_id){
// SimpleWebRTC (swr) is initialized
swr.joinRoom(room);
swr.once('joinedRoom', function(){
// the client joined the specified room
// initialize the connector with the required parameters.
// You always should specify `role`, `syncMethod`, and `user_id`
self.init({
role : "slave",
syncMethod : "syncAll",
user_id : user_id
});
var i;
// notify the connector class about all the users that already
// joined the session
for(i in self.swr.webrtc.peers){
self.userJoined(self.swr.webrtc.peers[i].id, "slave");
}
swr.on("channelMessage", function(peer, room, message){
// The client received a message
// Check if the connector is already initialized,
// only then forward the message to the connector class
if(self.is_initialized && message.type === "yjs"){
self.receiveMessage(peer.id, message.payload);
}
});
});
swr.on("createdPeer", function(peer){
// a new peer/client joined the session.
// Notify the connector class, if the connector
// is already initialized
if(self.is_initialized){
// note: Since the WebRTC Connector only supports the SyncAll
// syncmethod, every client is a slave.
self.userJoined(peer.id, "slave");
}
});
swr.on("peerStreamRemoved",function(peer){
// a client left the session.
// Notify the connector class, if the connector
// is already initialized
if(self.is_initialized){
self.userLeft(peer.id);
}
});
});
}
// Specify how to send a message to a specific user (by uid)
WebRTC.prototype.send = function(uid, message){
var self = this;
// we have to make sure that the message is sent under all circumstances
var send = function(){
// check if the clients still exists
var peer = self.swr.webrtc.getPeers(uid)[0];
var success;
if(peer){
// success is true, if the message is successfully sent
success = peer.sendDirectly("simplewebrtc", "yjs", message);
}
if(!success){
// resend the message if it didn't work
window.setTimeout(send,500);
}
};
// try to send the message
send();
};
// specify how to broadcast a message to all users
// (it may send the message back to itself).
// The webrtc connecor tries to send it to every single clients directly
WebRTC.prototype.broadcast = function(message){
this.swr.sendDirectlyToAll("simplewebrtc","yjs",message);
};
Y.Connectors.WebRTC = WebRTC;
})()
var connectorAdapter = (){
#
# @params new Connector(options)
# @param options.syncMethod {String} is either "syncAll" or "master-slave".
# @param options.role {String} The role of this client
# (slave or master (only used when syncMethod is master-slave))
# @param options.perform_send_again {Boolean} Whetehr to whether to resend the HB after some time period. This reduces sync errors, but has some overhead (optional)
#
init: (options)->
req = (name, choices)=>
if options[name]?
if (not choices?) or choices.some((c)->c is options[name])
@[name] = options[name]
else
throw new Error "You can set the '"+name+"' option to one of the following choices: "+JSON.encode(choices)
else
throw new Error "You must specify "+name+", when initializing the Connector!"
req "syncMethod", ["syncAll", "master-slave"]
req "role", ["master", "slave"]
req "user_id"
@on_user_id_set?(@user_id)
# whether to resend the HB after some time period. This reduces sync errors.
# But this is not necessary in the test-connector
if options.perform_send_again?
@perform_send_again = options.perform_send_again
else
@perform_send_again = true
# A Master should sync with everyone! TODO: really? - for now its safer this way!
if @role is "master"
@syncMethod = "syncAll"
# is set to true when this is synced with all other connections
@is_synced = false
# Peerjs Connections: key: conn-id, value: object
@connections = {}
# List of functions that shall process incoming data
@receive_handlers ?= []
# whether this instance is bound to any y instance
@connections = {}
@current_sync_target = null
@sent_hb_to_all_users = false
@is_initialized = true
onUserEvent: (f)->
@connections_listeners ?= []
@connections_listeners.push f
isRoleMaster: ->
@role is "master"
isRoleSlave: ->
@role is "slave"
findNewSyncTarget: ()->
@current_sync_target = null
if @syncMethod is "syncAll"
for user, c of @connections
if not c.is_synced
@performSync user
break
if not @current_sync_target?
@setStateSynced()
null
userLeft: (user)->
delete @connections[user]
@findNewSyncTarget()
if @connections_listeners?
for f in @connections_listeners
f {
action: "userLeft"
user: user
}
userJoined: (user, role)->
if not role?
throw new Error "Internal: You must specify the role of the joined user! E.g. userJoined('uid:3939','slave')"
# a user joined the room
@connections[user] ?= {}
@connections[user].is_synced = false
if (not @is_synced) or @syncMethod is "syncAll"
if @syncMethod is "syncAll"
@performSync user
else if role is "master"
# TODO: What if there are two masters? Prevent sending everything two times!
@performSyncWithMaster user
if @connections_listeners?
for f in @connections_listeners
f {
action: "userJoined"
user: user
role: role
}
#
# Execute a function _when_ we are connected. If not connected, wait until connected.
# @param f {Function} Will be executed on the Connector context.
#
whenSynced: (args)->
if args.constructor is Function
args = [args]
if @is_synced
args[0].apply this, args[1..]
else
@compute_when_synced ?= []
@compute_when_synced.push args
#
# Execute an function when a message is received.
# @param f {Function} Will be executed on the PeerJs-Connector context. f will be called with (sender_id, broadcast {true|false}, message).
#
onReceive: (f)->
@receive_handlers.push f
#
# perform a sync with a specific user.
#
performSync: (user)->
if not @current_sync_target?
@current_sync_target = user
@send user,
sync_step: "getHB"
send_again: "true"
data: @getStateVector()
if not @sent_hb_to_all_users
@sent_hb_to_all_users = true
hb = @getHB([]).hb
_hb = []
for o in hb
_hb.push o
if _hb.length > 10
@broadcast
sync_step: "applyHB_"
data: _hb
_hb = []
@broadcast
sync_step: "applyHB"
data: _hb
#
# When a master node joined the room, perform this sync with him. It will ask the master for the HB,
# and will broadcast his own HB
#
performSyncWithMaster: (user)->
@current_sync_target = user
@send user,
sync_step: "getHB"
send_again: "true"
data: @getStateVector()
hb = @getHB([]).hb
_hb = []
for o in hb
_hb.push o
if _hb.length > 10
@broadcast
sync_step: "applyHB_"
data: _hb
_hb = []
@broadcast
sync_step: "applyHB"
data: _hb
#
# You are sure that all clients are synced, call this function.
#
setStateSynced: ()->
if not @is_synced
@is_synced = true
if @compute_when_synced?
for el in @compute_when_synced
f = el[0]
args = el[1..]
f.apply(args)
delete @compute_when_synced
null
# executed when the a state_vector is received. listener will be called only once!
whenReceivedStateVector: (f)->
@when_received_state_vector_listeners ?= []
@when_received_state_vector_listeners.push f
#
# You received a raw message, and you know that it is intended for to Yjs. Then call this function.
#
receiveMessage: (sender, res)->
if not res.sync_step?
for f in @receive_handlers
f sender, res
else
if sender is @user_id
return
if res.sync_step is "getHB"
# call listeners
if @when_received_state_vector_listeners?
for f in @when_received_state_vector_listeners
f.call this, res.data
delete @when_received_state_vector_listeners
data = @getHB(res.data)
hb = data.hb
_hb = []
# always broadcast, when not synced.
# This reduces errors, when the clients goes offline prematurely.
# When this client only syncs to one other clients, but looses connectors,
# before syncing to the other clients, the online clients have different states.
# Since we do not want to perform regular syncs, this is a good alternative
if @is_synced
sendApplyHB = (m)=>
@send sender, m
else
sendApplyHB = (m)=>
@broadcast m
for o in hb
_hb.push o
if _hb.length > 10
sendApplyHB
sync_step: "applyHB_"
data: _hb
_hb = []
sendApplyHB
sync_step : "applyHB"
data: _hb
if res.send_again? and @perform_send_again
send_again = do (sv = data.state_vector)=>
()=>
hb = @getHB(sv).hb
for o in hb
_hb.push o
if _hb.length > 10
@send sender,
sync_step: "applyHB_"
data: _hb
_hb = []
@send sender,
sync_step: "applyHB",
data: _hb
sent_again: "true"
setTimeout send_again, 3000
else if res.sync_step is "applyHB"
@applyHB(res.data, sender is @current_sync_target)
if (@syncMethod is "syncAll" or res.sent_again?) and (not @is_synced) and ((@current_sync_target is sender) or (not @current_sync_target?))
@connections[sender].is_synced = true
@findNewSyncTarget()
else if res.sync_step is "applyHB_"
@applyHB(res.data, sender is @current_sync_target)
# Currently, the HB encodes operations as JSON. For the moment I want to keep it
# that way. Maybe we support encoding in the HB as XML in the future, but for now I don't want
# too much overhead. Y is very likely to get changed a lot in the future
#
# Because we don't want to encode JSON as string (with character escaping, wich makes it pretty much unreadable)
# we encode the JSON as XML.
#
# When the HB support encoding as XML, the format should look pretty much like this.
# does not support primitive values as array elements
# expects an ltx (less than xml) object
parseMessageFromXml: (m)->
parse_array = (node)->
for n in node.children
if n.getAttribute("isArray") is "true"
parse_array n
else
parse_object n
parse_object = (node)->
json = {}
for name, value of node.attrs
int = parseInt(value)
if isNaN(int) or (""+int) isnt value
json[name] = value
else
json[name] = int
for n in node.children
name = n.name
if n.getAttribute("isArray") is "true"
json[name] = parse_array n
else
json[name] = parse_object n
json
parse_object m
# encode message in xml
# we use string because Strophe only accepts an "xml-string"..
# So {a:4,b:{c:5}} will look like
# <y a="4">
# <b c="5"></b>
# </y>
# m - ltx element
# json - guess it ;)
#
encodeMessageToXml: (m, json)->
# attributes is optional
encode_object = (m, json)->
for name,value of json
if not value?
# nop
else if value.constructor is Object
encode_object m.c(name), value
else if value.constructor is Array
encode_array m.c(name), value
else
m.setAttribute(name,value)
m
encode_array = (m, array)->
m.setAttribute("isArray","true")
for e in array
if e.constructor is Object
encode_object m.c("array-element"), e
else
encode_array m.c("array-element"), e
m
if json.constructor is Object
encode_object m.c("y",{xmlns:"http://y.ninja/connector-stanza"}), json
else if json.constructor is Array
encode_array m.c("y",{xmlns:"http://y.ninja/connector-stanza"}), json
else
throw new Error "I can't encode this json!"
setIsBoundToY: ()->
@on_bound_to_y?()
delete @when_bound_to_y
@is_bound_to_y = true
}
};

46
src/OperationStore.js
View File

@ -26,6 +26,9 @@ type Id = [string, number];
class AbstractOperationStore { //eslint-disable-line no-unused-vars
constructor () {
this.parentListeners = {};
this.parentListenersRequestPending = false;
this.parentListenersActivated = {};
// E.g. this.listenersById[id] : Array<Listener>
this.listenersById = {};
// Execute the next time a transaction is requested
@ -106,6 +109,7 @@ class AbstractOperationStore { //eslint-disable-line no-unused-vars
}
// called by a transaction when an operation is added
operationAdded (op) {
// notify whenOperation listeners (by id)
var l = this.listenersById[op.id];
if (l != null) {
for (var listener of l){
@ -114,5 +118,47 @@ class AbstractOperationStore { //eslint-disable-line no-unused-vars
}
}
}
// notify parent listeners, if possible
var listeners = this.parentListeners[op.parent];
if ( this.parentListenersRequestPending
|| ( listeners == null )
|| ( listeners.length === 0 )) {
return;
}
var al = this.parentListenersActivated[JSON.stringify(op.parent)];
if ( al == null ){
al = [];
this.parentListenersActivated[JSON.stringify(op.parent)] = al;
}
al.push(op);
this.parentListenersRequestPending = true;
var store = this;
this.requestTransaction(function*(myRequest){ // you can throw error on myRequest, then restart if you have to
store.parentListenersRequestPending = false;
var activatedOperations = store.parentListenersActivated;
store.parentListenersActivated = {};
for (var parent_id in activatedOperations){
var parent = yield* this.getOperation(parent_id);
Struct[parent.type].notifyObservers(activatedOperations[parent_id]);
}
})
}
removeParentListener (id, f) {
var ls = this.parentListeners[id];
if (ls != null) {
this.parentListeners[id] = ls.filter(function(g){
return (f !== g);
});
}
}
addParentListener (id, f) {
var ls = this.parentListeners[JSON.stringify(id)];
if (ls == null) {
ls = [];
this.parentListeners[JSON.stringify(id)] = ls;
}
ls.push(f);
}
}

2
src/OperationStores/IndexedDB.js
View File

@ -128,7 +128,7 @@ var IndexedDB = (function(){ //eslint-disable-line no-unused-vars
var request = yield transactionQueue;
transaction = new Transaction(store);
yield* request.call(transaction);/*
yield* request.call(transaction, request);/*
while (transactionQueue.queue.length > 0) {
yield* transactionQueue.queue.shift().call(transaction);
}*/

116
src/Operations.js
View File

@ -1,7 +1,7 @@
/* @flow */
// Op is anything that we could get from the OperationStore.
type Op = Object;
struct Op = Object;
var Struct = {
Operation: { //eslint-disable-line no-unused-vars
@ -14,13 +14,15 @@ var Struct = {
Insert: {
create: function*( op : Op,
user : string,
content : any,
left : Struct.Insert,
right : Struct.Insert) : Struct.Insert {
right : Struct.Insert,
parent : Struct.List) : Struct.Insert {
op.left = left ? left.id : null;
op.origin = op.left;
op.right = right ? right.id : null;
op.type = "Insert";
yield* Struct.Operation.create(op, user);
op.struct = "Insert";
yield* Struct.Operation.create.call(this, op, user);
if (left != null) {
left.right = op.id;
@ -33,12 +35,112 @@ var Struct = {
return op;
},
requiredOps: function(op, ids){
ids.push(op.left);
ids.push(op.right);
if(op.left != null){
ids.push(op.left);
}
if(op.right != null){
ids.push(op.right);
}
return ids;
},
getDistanceToOrigin: function *(op){
var d = 0;
var o = yield this.getOperation(op.left);
while (op.origin !== (o ? o.id : null)) {
d++;
o = yield this.getOperation(o.left);
}
return d;
},
/*
# $this has to find a unique position between origin and the next known character
# case 1: $origin equals $o.origin: the $creator parameter decides if left or right
# let $OL= [o1,o2,o3,o4], whereby $this is to be inserted between o1 and o4
# o2,o3 and o4 origin is 1 (the position of o2)
# there is the case that $this.creator < o2.creator, but o3.creator < $this.creator
# then o2 knows o3. Since on another client $OL could be [o1,o3,o4] the problem is complex
# therefore $this would be always to the right of o3
# case 2: $origin < $o.origin
# if current $this insert_position > $o origin: $this ins
# else $insert_position will not change
# (maybe we encounter case 1 later, then this will be to the right of $o)
# case 3: $origin > $o.origin
# $this insert_position is to the left of $o (forever!)
*/
execute: function*(op){
return op;
var distance_to_origin = yield* Struct.Insert.getDistanceToOrigin(op); // most cases: 0 (starts from 0)
var i = distance_to_origin; // loop counter
var o = yield* this.getOperation(this.left);
o = yield* this.getOperation(o.right);
var tmp;
while (true) {
if (o.id !== this.right){
if (Struct.Insert.getDistanceToOrigin(o) === i) {
// case 1
if (o.id[0] < op.id[0]) {
op.left = o;
distance_to_origin = i + 1;
}
} else if ((tmp = Struct.Insert.getDistanceToOrigin(o)) < i) {
// case 2
if (i - distance_to_origin <= tmp) {
op.left = o;
distance_to_origin = i+1;
}
} else {
break;
}
i++;
o = yield* this.getOperation(o.next_cl);
} else {
break
}
}
// reconnect..
var left = this.getOperation(op.left);
var right = this.getOperation(op.right);
left.right = op.id;
right.left = op.id;
op.left = left;
op.right = right;
yield* this.setOperation(left);
yield* this.setOperation(right);
yield* this.setOperation(op);
}
},
List: {
create: function*( op : Op,
user : string){
op.start = null;
op.end = null;
op.struct = "List";
return yield* Struct.Operation.create.call(this, op, user);
},
requiredOps: function(op, ids){
if (op.start != null) {
ids.push(op.start);
}
if (op.end != null){
ids.push(op.end);
}
return ids;
},
execute: function* (op) {
// nop
}
ref: function* (op, pos) : Struct.Insert | undefined{
var o = op.start;
while ( pos !== 0 || o == null) {
o = (yield* this.getOperation(op.start)).right;
}
return (o == null) ? null : yield* this.getOperation(o);
}
insert: function* (op, pos : number, contents : Array<any>) {
var o = yield* Struct.List.ref.call(this, op, pos);
var o_end = yield* this.getOperation(o.right);
for (var content of contents) {
o = yield* Struct.Insert.create.call(this, {}, user, content, o, o_end, op);
}
}
}
};

25
src/Types.js Normal file
View File

@ -0,0 +1,25 @@
(function(){
class List {
constructor (_model) {
this._model = _model;
}
*val (pos) {
var o = yield* this.Struct.List.ref(pos);
return o ? o.content : null;
}
*insert (pos, contents) {
yield* this.Struct.List.insert(pos, contents);
}
}
Y.List = function* YList(){
var model = yield* this.Struct.List.create();
return new Y.List.Create(model);
}
Y.List.Create = List;
Y.List = List;
})();

5
src/y.js
View File

@ -1 +1,6 @@
/* @flow */
function Y (opts) {
var connector = opts.connector;
Y.Connectors[connector.name]
}

2
y.js
View File

File diff suppressed because one or more lines are too long

2
y.js.map
View File

File diff suppressed because one or more lines are too long