L2l

channel.js

@@ -1,5 +1,4 @@
 import { string, num, promise, fun } from "lively.lang";
-
 var debug = false;
 
 // -=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
@@ -24,8 +23,13 @@ export class Channel {
     this.delayBtoA = 0;
     this.online = false;
     this.lifetime = 100;
-    this._watchdogProcess = null
-    this.goOnline();
+    this._watchdogProcess = null    
+  }
+
+  static establish(senderRecvrA, onReceivedMethodA, senderRecvrB, onReceivedMethodB){
+    var channel = new this(senderRecvrA, onReceivedMethodA, senderRecvrB, onReceivedMethodB)
+    channel.goOnline();
+    return channel;
   }
 
   toString() {
@@ -36,6 +40,13 @@ export class Channel {
   goOffline() { this.online = false; }
   goOnline() { this.online = true; this.watchdogProcess(); }
 
+  whenOnline(timeout) {
+    return promise.waitFor(timeout, () => this.isOnline())
+            .catch(err =>
+              Promise.reject(/timeout/i.test(String(err)) ?
+                new Error(`Timeout in ${this}.whenOnline`) : err))
+  }
+
   watchdogProcess() {
     if (!this.isOnline() || this._watchdogProcess) return;
 
@@ -98,17 +109,24 @@ export class Channel {
       Promise.resolve().then(() => {
         if (!delay) {
           var outgoing = queue.slice(); queue.length = 0;
-          try { recvr[method](outgoing, sender, this); }
+          try { this.doSend(sender,outgoing,this);}
           catch (e) { console.error(`Error in ${method} of ${recvr}: ${e.stack || e}`); }
         } else {
           fun.throttleNamed(`${this.id}-${descr}`, delay*1000, () => {
             var outgoing = queue.slice(); queue.length = 0;
-            try { recvr[method](outgoing, sender, this); }
+            try { this.doSend(sender,outgoing,this);}
             catch (e) { console.error(`Error in ${method} of ${recvr}: ${e.stack || e}`); }
           })();
         }
       });
 
     return promise.waitFor(() => queue.length === 0);
   }
-}
+
+  doSend(sender,outgoing,self) {
+    var { recvr, method, queue, delay, descr } = this.componentsForSender(sender); 
+    recvr[method](outgoing,sender,self)
+  }  
+}
+
+

client.js

@@ -2,8 +2,10 @@ import { obj, arr, string, promise } from "lively.lang";
 import { morphicDefaultTransform, transformOp_1_to_n, composeOps } from "./transform.js";
 import { printOps, printOp } from "./debugging.js";
 import { Channel } from "./channel.js";
+// import { L2LChannel as Channel} from "./l2lchannel.js";
 import { serializeChange, applyChange } from "./changes.js";
 
+
 var i = val => obj.inspect(val, {maxDepth: 2});
 
 export function fnStringsToFunctions(fnStrings) {
@@ -88,15 +90,18 @@ export class Client {
     metaChannel && metaChannel.goOffline();
     var master = (opChannel || metaChannel).senderRecvrB;
     master.removeConnection(con);
+    console.log(con.metaChannel)
     con.metaChannel = null;
     con.opChannel = null;
   }
 
-  connectToMaster(master) {
+  async connectToMaster(master) {
     this.disconnectFromMaster();
     var con = this.state.connection;
-    con.opChannel = new Channel(this, "receiveOpsFromMaster", master, "receiveOpsFromClient")
-    con.metaChannel = new Channel(this, "receiveMetaMsgsFromMaster", master, "receiveMetaMsgsFromClient")
+    con.opChannel = Channel.establish(this, "receiveOpsFromMaster", master, "receiveOpsFromClient")
+    await con.opChannel.whenOnline(500)
+    con.metaChannel = Channel.establish(this, "receiveMetaMsgsFromMaster", master, "receiveMetaMsgsFromClient")
+    await con.metaChannel.whenOnline(500)
     master.addConnection(con);
   }
 

l2lchannel.js

@@ -0,0 +1,92 @@
+import L2LClient from "lively.2lively/client.js";
+import { string, num, promise, fun } from "lively.lang";
+import { Channel } from "./channel.js";
+
+
+export class L2LChannel extends Channel{
+  constructor(senderRecvrA, onReceivedMethodA, senderRecvrB, onReceivedMethodB){     
+      super(senderRecvrA, onReceivedMethodA, senderRecvrB, onReceivedMethodB);
+      if (!this.senderRecvrA.l2lclient){ this.senderRecvrA.l2lclient = L2LChannel.makeL2LClient() }
+      if (!this.senderRecvrB.l2lclient){ this.senderRecvrB.l2lclient = L2LChannel.makeL2LClient() }      
+  }  
+
+  static makeL2LClient(hostname,port, namespace,io){
+    var client = L2LClient.forceNew({})
+    return client
+  }
+
+  registerReceive(client,method){
+    var l2l = client.l2lclient;
+    l2l.addService("lively.sync", (tracker, msg, ackFn, sender) => {
+        method(msg.data)
+    });
+  }
+
+  async getSessions(senderA,senderB,ackFn){    
+    var l2lB = senderB.l2lclient    
+    l2lB.sendTo(l2lB.trackerId,'listRoom',{roomName: l2lB.socketId.split('#')[1]},(a) => {ackFn(a)})    
+  }
+
+  isOnline() { return this.online; }
+
+  goOffline() {
+    console.log(this)
+    this.online = false;
+
+    var l2lA = this.senderRecvrA.l2lclient,
+    l2lB = this.senderRecvrB.l2lclient
+    if(!l2lA || l2lA.isRegistered() == false){ console.log('l2lA Missing');return}
+    if(!l2lB || l2lB.isRegistered() == false){ console.log('l2lB Missing');return}
+    l2lA.sendTo(l2lA.trackerId,'leaveRoom',{roomName: l2lB.socketId.split('#')[1]})
+
+    if (this.senderRecvrA && this.senderRecvrA.l2lclient) {
+      this.senderRecvrA.l2lclient.remove()
+      console.log('senderRecvrA disconnected')
+
+    } else {      
+      console.log('senderRecvrA not disconnected');      
+    }
+
+    if (this.senderRecvrB && this.senderRecvrB.l2lclient) {
+
+      this.getSessions(this.senderRecvrA,this.senderRecvrB,(a)=> {
+        if(a.data.sockets.hasOwnProperty(l2lB.socketId)){
+          l2lB.sendTo(l2lB.trackerId,'leaveRoom',{roomName: l2lB.socketId.split('#')[1]})  
+          l2lB.remove();          
+        } else {
+          console.log('senderRecvrB not disconnected');
+        }    
+      })
+    }
+
+  }
+
+
+  doSend(sender,outgoing,self) {
+    var { recvr, method, queue, delay, descr } = this.componentsForSender(sender); 
+    sender.l2lclient.sendTo(recvr.l2lclient.id,'lively.sync',outgoing,()=>{console.log('received')})
+    // recvr[method](outgoing,sender,self)
+  }
+
+
+
+  async goOnline() {
+    var l2lA = this.senderRecvrA.l2lclient,
+    l2lB = this.senderRecvrB.l2lclient
+    await l2lA.whenRegistered(300)
+    await l2lB.whenRegistered(300)
+
+    this.registerReceive(this.senderRecvrA,this.senderRecvrA.receiveOpsFromMaster)
+    this.registerReceive(this.senderRecvrB,this.senderRecvrB.receiveOpsFromClient)
+
+    l2lA.sendTo(l2lA.trackerId,'joinRoom',{roomName: l2lB.socketId.split('#')[1]})
+    l2lB.sendTo(l2lB.trackerId,'joinRoom',{roomName: l2lB.socketId.split('#')[1]})
+    this.online = true;
+    this.watchdogProcess();
+  }
+
+
+
+
+
+}

tests/sync-test.js

@@ -8,7 +8,10 @@ import { disconnect, disconnectAll, connect } from "lively.bindings";
 import { buildTestWorld, destroyTestWorld } from "./helper.js";
 import { Client } from "../client.js";
 import { Master } from "../master.js";
-
+import { L2LChannel as L2LChannel } from "../l2lchannel.js";
+import L2LClient from "lively.2lively/client.js";
+import L2LTracker from "lively.2lively/tracker.js";
+import { Channel} from "../channel.js";
 
 // System.decanonicalize("mocha-es6", "http://localhost:9011/lively.sync/tests/sync-test.js")
 // var env1, env2, env3,
@@ -27,7 +30,7 @@ async function setup(nClients) {
   for (var i = 0; i < nClients; i++) {
     let env = state[`env${i+1}`] = await buildTestWorld(masterEnv.world.exportToJSON(), pt(0,300*(i+1))),
         client = state[`client${i+1}`] = new Client(env.world, `client${i+1}`);
-    client.connectToMaster(master);
+    await client.connectToMaster(master);
     state[`world${i+1}`] = env.world;
     connect(env.changeManager, 'changeRecorded', client, 'newChange');
   }
@@ -85,6 +88,102 @@ describe("messaging between master and client", () => {
 
 });
 
+
+
+describe("lively2lively backchannel tests", function() {
+  this.timeout(1*1000)
+  before(async() =>{
+    testingClient = await L2LClient.forceNew({});
+    await testingClient.whenRegistered(300);
+  })
+  after(async() => {
+    await testingClient.remove();
+  })
+  beforeEach(async () => setup(2));
+  afterEach(async () => teardown());
+
+  it("ensure clients have l2l clients", async () => {
+
+    var {world1, masterWorld, client1, master} = state;    
+    var testChannel = L2LChannel.establish(client1, "receiveOpsFromMaster", master, "receiveOpsFromClient")
+
+    expect((testChannel.senderRecvrA.l2lclient) &&  (testChannel.senderRecvrA.l2lclient instanceof L2LClient)).equals(true,'client A not L2LClient')
+    expect((testChannel.senderRecvrB.l2lclient) &&  (testChannel.senderRecvrB.l2lclient instanceof L2LClient)).equals(true,'client B not L2LClient')
+
+    await testChannel.whenOnline(300)
+
+    testChannel.goOffline();
+  })
+
+  it('ensure clients can share a master', async () => {
+
+    var {world1, masterWorld, client1, client2, master} = state;    
+    var testChannel = L2LChannel.establish(client1, "receiveOpsFromMaster", master, "receiveOpsFromClient")    
+    window.testChannel = testChannel
+    expect((testChannel.senderRecvrA.l2lclient) &&  (testChannel.senderRecvrA.l2lclient instanceof L2LClient)).equals(true,'client A not L2LClient')
+    expect((testChannel.senderRecvrB.l2lclient) &&  (testChannel.senderRecvrB.l2lclient instanceof L2LClient)).equals(true,'client B not L2LClient')
+
+    await testChannel.whenOnline(300)
+
+    var channelId = testChannel.senderRecvrB.l2lclient.socketId.split('#')[1]
+
+    var testChannel2 = L2LChannel.establish(client2, "receiveOpsFromMaster", master, "receiveOpsFromClient")    
+
+    expect((testChannel2.senderRecvrA.l2lclient) &&  (testChannel2.senderRecvrA.l2lclient instanceof L2LClient)).equals(true,'client A not L2LClient')
+    expect((testChannel2.senderRecvrB.l2lclient) &&  (testChannel2.senderRecvrB.l2lclient instanceof L2LClient)).equals(true,'client B not L2LClient')
+
+    await testChannel2.whenOnline(300)
+
+    expect(testChannel2.senderRecvrB === testChannel.senderRecvrB).equals(true,'Clients have different masters')
+
+    testChannel.goOffline();
+    testChannel2.goOffline();
+  })
+
+  it('ensure clients can communicate across channel', async () => {
+    var {world1, masterWorld, client1, master} = state;  
+    var client1Buffer = []
+    var masterBuffer = []
+    var payload = "Test for lively.sync messaging"
+    var ack = 'received'
+    function masterCheck(item){      
+      masterBuffer.push(item)
+    }
+    function ackFn(item){
+        client1Buffer.push(item)        
+        client1.l2lclient.sendTo(master.l2lclient.id,'lively.sync',{payload: ack})
+    }
+    client1.receiveOpsFromMaster = ackFn;
+    master.receiveOpsFromClient = masterCheck;
+
+    var testChannel = L2LChannel.establish(client1, "receiveOpsFromMaster", master, "receiveOpsFromClient")
+
+    await testChannel.whenOnline(500)
+
+    master.l2lclient.sendTo(client1.l2lclient.id,'lively.sync',{payload: payload})
+    await promise.waitFor(200, () => client1Buffer.length  >= 1);
+    await promise.waitFor(200, () => masterBuffer.length >= 1);
+    testChannel.goOffline();
+    expect(client1Buffer[0].payload).equals(payload,'payload not correct')
+    expect(masterBuffer[0].payload).equals(ack,'payload not correct')
+  })
+
+  it('Temp test, check status of inherited methods',async() => {
+
+    var {world1, masterWorld, client1, client2, master} = state;    
+    var opChannel = Channel.establish(client1, "receiveOpsFromMaster", master, "receiveOpsFromClient")
+    // await opChannel.whenOnline(300)
+    var metaChannel = L2LChannel.establish(client1, "receiveOpsFromMaster", master, "receiveOpsFromClient")
+    await metaChannel.whenOnline(300)
+    console.log({state,opChannel,metaChannel,client1,master})
+    opChannel.goOffline()
+    metaChannel.goOffline()
+
+  })
+
+})
+
+
 describe("syncing master with two clients", function() {
 
   // this.timeout(5*1000);
@@ -192,7 +291,7 @@ describe("syncing master with two clients", function() {
     expect(masterWorld.fill).equals(Color.green);
     expect(world1.fill).equals(Color.yellow);
 
-    client1.connectToMaster(master);
+    await client1.connectToMaster(master);
     await client1.syncWithMaster();
     expect(world1.fill).equals(Color.green);
   });