Update pseudocode

Cache

-class Cache {
-
-  private cobjects: Map<CObjectId, CObject>
-  
-  fun getObject(oid: CObjectId) {
-    return cobjects.get(oid)
-  }
-
-  fun addObject(oid: CObjectId, obj: CObject) {
-    this.cobjects.put(oid. obj)
-  }
-}

Collection

 class Collection {
 
+  // The collection id
   private id: CollectionId
 
+  // Is the collection open in read only mode
   private readOnly: Boolean
 
+  // Static boolean used to ensure that only one collection is openned
   private static otherOpenned: Boolean = false
 
   Collection(cid: CollectionId) {
@@ -12,21 +15,26 @@ class Collection {
     this.id = cid
   }
 
-  // c_open_read|write<T> handler?
-  // TODO: manage notifications
-  fun open<T>(oid: String, readOnly: Boolean, handler: (VersionVector, List<CObjectId>) -> Unit = {}): T {
+  // c_open_read|write<T>
+  fun open<T>(oid: String, readOnly: Boolean, handler: NotificationHandler): T {
+    if (this.readOnly AND NOT readOnly) throw Exception
     oid = CObjectId(this.id, oid, T)
     if (Global.Cache.contains(oid)) return Global.Cache.get(oid)
-    crdt = SERVICE.getObject(oid, TxnEnv.getState()) // SERVICE.subscribe(oid)
-    if (this.readOnly AND NOT readOnly) throw Exception
-    return CObject<T>(oid, crdt, readOnly)
+    crdt, _ = SERVICE.getObject(oid, TxnEnv.getState())
+    SERVICE.subscribe(oid)
+    obj = CObject<T>(oid, crdt, readOnly)
+    Global.Cache.put(oid, obj)
+    Global.Handlers.put(oid, handler)
+    return obj
   }
 
+  // c_close_collection
   fun close() {
-    for obj in Global.Cache {
-      SERVICE.unsubscribe(obj.id)
+    for oid in Global.Cache {
+      SERVICE.unsubscribe(oid)
     }
     Global.Cache.clear()
+    Global.Handlers.clear()
     Collection.otherOpenned = false
   }
 }

Global

+// Transaction environment should inherit from crdtlib.utils.Environment 
 TxnEnv: TransactionEnvironment
 
+// Map for dirty objects
 DirtyCObjects: Map<CObjectId, CObject>
 
+// Map for the cache
 Cache: Map<CObjectId, CObject>
+
+// Map for c-object handlers management
+Handlers: Map<CObjectId, NotificationHandler>

NotificationHandler

+// The notification handler (function) type accepted for c-objects
+type NotificationHandler = (VersionVector, CObjectId) -> Unit

ServiceHandler

+// Thread that will listen for server notifications
+class ServiceHandler: Thread {
+
+  fun run() {
+    this.run(work)
+  }
+
+  fun work() {
+    while (true) {
+      v, oid = readMessage()
+      if (Global.Handlers.contains(oid)) {
+        // call the corresponding handler
+        Handlers.get(oid)(v, oid)
+      }
+    }
+  }
+
+  fun stop() {
+    exit(this)
+  }
+}

Session

 class Session {
 
+  // Current consistency level
   private consistency: ConsistencyLevel
 
+  // The client id
   private cliendId: CliendId
 
+  // The attached service handler
+  private serviceHandler: ServiceHandler
+
   Session(cid: ClientId) {
     this.clientId = cid
   }
@@ -12,6 +17,8 @@ class Session {
   static fun open(dbName: String, credentials: String): Session {
     clientId = ClientId()
     if (NOT SERVICE.open(dbName, credentials, clientId)) throw Exception
+    this.serviceHandler = ServiceHandler()
+    this.serviceHandler.run()
     return Session(clientId)
   }
 
@@ -19,16 +26,24 @@ class Session {
   fun pull(type: ConsistencyLevel) {
     if (consistency != null AND consistency > type) throw Exception
     this.consistency = type
-    v = SERVICE.pull(type)
-    Global.TxnEnv.updateState(v)
+    for oid in Cache {
+      crdt, v = SERVICE.getObject(oid)
+      obj = Global.Cache.get(oid)
+      obj.crdt = crdt
+      Global.TxnEnv.updateSate(v)
+    } 
   }
 
   // c_pull_XX_view(v)
   fun pull(type: ConsistencyLevel, vv: VersionVector) {
     if (view != null AND view.type > type) throw Exception
     if (vv < global.TxnEnv.getState()) throw Exception
-    v = SERVICE.pull(type, vv)
-    Global.TxnEnv.updateState(v)
+    for oid in Cache {
+      crdt, v = SERVICE.getObject(oid, vv)
+      obj = Global.Cache.get(oid)
+      obj.crdt = crdt
+      Global.TxnEnv.updateSate(v)
+    } 
   }
   
   // c_open_collection_read|write
@@ -44,5 +59,6 @@ class Session {
   // c_end_session
   fun close() {
     SERVICE.close(clientId)
+    this.serviceHandler.stop()
   }
 }

Transaction

 class Transaction {
 
-  private transactionId: String
+  // Transaction id
+  private transactionId: TxnId
 
+  // Static boolean used to ensure that only one collection is openned
   private static otherOpenned: Boolean = false
 
+  // The version vector at which the transaction has started
   private begin: VersionVector
 
-  Transaction(tid: TxnId, body: () -> Unit) {
+  Transaction(tid: TxnId, body: TransactionBody) {
     if (Transaction.otherOpenned) throw Exception
     Transaction.otherOpenned = true
     this.begin = Global.TxnEnv.getState()
     this.transactionId = Global.TxnEnv.tickTransaction()
 
     try {
+      // call the body function
       body()
+      // if everything goes well commit
       this.commit()
     } catch {
+      // If there is an exception abort
       this.abort()
     }
   }
@@ -33,7 +39,7 @@ class Transaction {
   fun commit() {
     SERVICE.beginTransaction(this.transactionId)
     for obj in Global.DirtyCObjects {
-      SERVICE.updateObject(obj.crdt.getDelta(this.begin))
+      SERVICE.pushObject(obj.crdt.getDelta(this.begin))
       obj.txnCommit()
     }
     SERVICE.commitTransaction(this.transactionId)

TransactionBody

+// The function type accepted for a transaction body
+type TransactionBody = () -> Unit