Merge remote-tracking branch 'origin/refactor2'

examples/keyvalue.js

@@ -40,7 +40,6 @@ let run = (async(() => {
     }
 
   } catch(e) {
-    console.error("error:", e);
     console.error(e.stack);
     process.exit(1);
   }

src/OrbitClient.js

@@ -4,55 +4,43 @@ const EventEmitter = require('events').EventEmitter;
 const async        = require('asyncawait/async');
 const await        = require('asyncawait/await');
 const ipfsDaemon   = require('orbit-common/lib/ipfs-daemon');
-const ipfsAPI      = require('orbit-common/lib/ipfs-api-promised');
-const Operations   = require('./list/Operations');
-const List         = require('./list/OrbitList');
-const OrbitDBItem  = require('./db/OrbitDBItem');
-const ItemTypes    = require('./db/ItemTypes');
-const MetaInfo     = require('./db/MetaInfo');
-const Post         = require('./db/Post');
 const PubSub       = require('./PubSub');
+const OrbitDB      = require('./OrbitDB');
 
-class OrbitDB {
+class OrbitClient {
   constructor(ipfs, daemon) {
     this._ipfs = ipfs;
-    this._store = {};
     this._pubsub = null;
     this.user = null;
     this.network = null;
     this.events = new EventEmitter();
+    this.db = new OrbitDB(this._ipfs);
   }
 
-  channel(hash, password, subscribe) {
+  channel(channel, password, subscribe) {
     if(password === undefined) password = '';
     if(subscribe === undefined) subscribe = true;
 
-    this._store[hash] = new List(this._ipfs, this.user.username);
+    this.db.use(channel, this.user, password);
-
+    this.db.events.on('data', async((hash) => {
-    const onMessage = async((hash, message) => {
+      await(this._pubsub.publish(channel, hash));
-      // console.log("--> Head:", message)
+      this.events.emit('data', channel, hash);
-      if(message && this._store[hash]) {
+    }));
-        const other = List.fromIpfsHash(this._ipfs, message);
-        this._store[hash].join(other);
-      }
-      this.events.emit('data', hash, message);
-    });
 
     if(subscribe)
-      this._pubsub.subscribe(hash, password, onMessage, onMessage);
+      this._pubsub.subscribe(channel, password, async((channel, message) => this.db.sync(channel, message)));
 
     return {
-      iterator: (options) => this._iterator(hash, password, options),
+      iterator: (options) => this._iterator(channel, password, options),
-      delete: () => this._deleteChannel(hash, password),
+      delete: () => this.db.deleteChannel(channel, password),
-      add: (data) => this._add(hash, password, data),
+      del: (key) => this.db.del(channel, password, key),
-      del: (key) => this._remove(hash, password, key),
+      add: (data) => this.db.add(channel, password, data),
-      put: (key, data) => this._put(hash, password, key, data),
+      put: (key, data) => this.db.put(channel, password, key, data),
       get: (key, options) => {
-        const items = this._iterator(hash, password, { key: key }).collect();
+        const items = this._iterator(channel, password, { key: key }).collect();
         return items[0] ? items[0].payload.value : null;
       },
-      //TODO: tests
+      leave: () => this._pubsub.unsubscribe(channel)
-      leave: () => this._pubsub.unsubscribe(hash)
     }
   }
 
@@ -61,10 +49,11 @@ class OrbitDB {
     this._store = {};
     this.user = null;
     this.network = null;
+    this.db = null;
   }
 
   _iterator(channel, password, options) {
-    const messages = this._getMessages(channel, password, options);
+    const messages = this.db.read(channel, password, options);
     let currentIndex = 0;
     let iterator = {
       [Symbol.iterator]() {
@@ -84,67 +73,6 @@ class OrbitDB {
     return iterator;
   }
 
-  _getMessages(channel, password, options) {
-    let opts = options || {};
-    Object.assign(opts, { amount: opts.limit || 1 });
-    let messages = await(this._store[channel].findAll(opts));
-    if(opts.reverse) messages.reverse();
-    return messages;
-  }
-
-  _publish(data) {
-    return new Promise((resolve, reject) => {
-      let post = new Post(data);
-      // post.encrypt(privkey, pubkey);
-      const res = await (ipfsAPI.putObject(this._ipfs, JSON.stringify(post)));
-      resolve(res);
-    })
-  }
-
-  _createMessage(channel, password, operation, key, value) {
-    const size = -1;
-    const meta = new MetaInfo(ItemTypes.Message, size, this.user.username, new Date().getTime());
-    const item = new OrbitDBItem(operation, key, value, meta);
-    const data = await (ipfsAPI.putObject(this._ipfs, JSON.stringify(item)));
-    return data.Hash;
-  }
-
-  /* DB Operations */
-  _add(channel, password, data) {
-    const post = await(this._publish(data));
-    const key = post.Hash;
-    return await(this._createOperation(channel, password, Operations.Add, key, post.Hash, data));
-  }
-
-  _put(channel, password, key, data) {
-    const post = await(this._publish(data));
-    return await(this._createOperation(channel, password, Operations.Put, key, post.Hash));
-  }
-
-  _remove(channel, password, hash) {
-    return await(this._createOperation(channel, password, Operations.Delete, hash, null));
-  }
-
-  _createOperation(channel, password, operation, key, value, data) {
-    var createOperation = async(() => {
-      return new Promise(async((resolve, reject) => {
-        const hash = this._createMessage(channel, password, operation, key, value);
-        const res = await(this._store[channel].add(hash));
-        const listHash = await(this._store[channel].ipfsHash);
-        await(this._pubsub.publish(channel, listHash));
-        resolve();
-      }));
-    })
-    await(createOperation());
-    return key;
-    // return res;
-  }
-
-  _deleteChannel(channel, password) {
-    this._store[channel].clear();
-    return true;
-  }
-
   _connect(host, port, username, password, allowOffline) {
     if(allowOffline === undefined) allowOffline = false;
     try {
@@ -165,7 +93,7 @@ class OrbitClientFactory {
       ipfs = ipfsd.ipfs;
     }
 
-    const client = new OrbitDB(ipfs);
+    const client = new OrbitClient(ipfs);
     await(client._connect(host, port, username, password, allowOffline))
     return client;
   }

src/OrbitDB.js

@@ -0,0 +1,98 @@
+'use strict';
+
+const EventEmitter = require('events').EventEmitter;
+const async        = require('asyncawait/async');
+const await        = require('asyncawait/await');
+const ipfsAPI      = require('orbit-common/lib/ipfs-api-promised');
+const Operations   = require('./list/Operations');
+const List         = require('./list/OrbitList');
+const OrbitDBItem  = require('./db/OrbitDBItem');
+const ItemTypes    = require('./db/ItemTypes');
+const MetaInfo     = require('./db/MetaInfo');
+const Post         = require('./db/Post');
+
+class OrbitDB {
+  constructor(ipfs) {
+    this._ipfs = ipfs;
+    this._logs = {};
+    this.events = new EventEmitter();
+  }
+
+  /* Public methods */
+  use(channel, user, password) {
+    this.user = user;
+    this._logs[channel] = new List(this._ipfs, this.user.username);
+  }
+
+  sync(channel, hash) {
+    console.log("--> Head:", hash)
+    if(hash && this._logs[channel]) {
+      const other = List.fromIpfsHash(this._ipfs, hash);
+      this._logs[channel].join(other);
+    }
+  }
+
+  /* DB Operations */
+  read(channel, password, options) {
+    let opts = options || {};
+    Object.assign(opts, { amount: opts.limit || 1 });
+    let messages = await(this._logs[channel].find(opts));
+    if(opts.reverse) messages.reverse();
+    return messages;
+  }
+
+  add(channel, password, data) {
+    const post = await(this._publish(data));
+    const key = post.Hash;
+    return await(this._createOperation(channel, password, Operations.Add, key, post.Hash, data));
+  }
+
+  put(channel, password, key, data) {
+    const post = await(this._publish(data));
+    return await(this._createOperation(channel, password, Operations.Put, key, post.Hash));
+  }
+
+  del(channel, password, hash) {
+    return await(this._createOperation(channel, password, Operations.Delete, hash, null));
+  }
+
+  deleteChannel(channel, password) {
+    this._logs[channel].clear();
+    return true;
+  }
+
+  /* Private methods */
+  _createOperation(channel, password, operation, key, value, data) {
+    var createOperation = async(() => {
+      return new Promise(async((resolve, reject) => {
+        const hash = this._createMessage(channel, password, operation, key, value);
+        const res = await(this._logs[channel].add(hash));
+        const listHash = await(this._logs[channel].ipfsHash);
+        resolve(listHash);
+      }));
+    })
+    const hash = await(createOperation());
+    this.events.emit('data', hash);
+    return key;
+  }
+
+  _createMessage(channel, password, operation, key, value) {
+    const size = -1;
+    const meta = new MetaInfo(ItemTypes.Message, size, this.user.username, new Date().getTime());
+    const item = new OrbitDBItem(operation, key, value, meta);
+    const data = await (ipfsAPI.putObject(this._ipfs, JSON.stringify(item)));
+    return data.Hash;
+  }
+
+  _publish(data) {
+    return new Promise((resolve, reject) => {
+      let post = new Post(data);
+      // post.encrypt(privkey, pubkey);
+      const res = await (ipfsAPI.putObject(this._ipfs, JSON.stringify(post)));
+      resolve(res);
+    })
+  }
+
+}
+
+module.exports = OrbitDB;

src/PubSub.js

@@ -1,7 +1,6 @@
 'use strict';
 
 const io   = require('socket.io-client');
-const List = require('./list/OrbitList');
 
 class Pubsub {
   constructor(ipfs) {
@@ -16,22 +15,12 @@ class Pubsub {
       this._socket = io.connect(`http://${host}:${port}`, { 'forceNew': true });
       this._socket.on('connect', resolve);
       this._socket.on('connect_error', (err) => reject(new Error(`Connection refused to ${host}:${port}`)));
-
+      this._socket.on('disconnect', (socket) => console.log(`Disconnected from http://${host}:${port}`));
-      // TODO: cleanup
+      this._socket.on('error', (e) => console.log('Pubsub socket error:', e));
-      this._socket.on('disconnect', (socket) => {
-        // console.log(`Disconnected from http://${host}:${port}`)
-      });
-
-      this._socket.on('error', (e) => console.log('error:', e));
       this._socket.on('message', this._handleMessage.bind(this));
       this._socket.on('latest', (hash, message) => {
         console.log(">", hash, message);
-        if(this._subscriptions[hash]) {
+        this._handleMessage(hash, message);
-          this._subscriptions[hash].head = message;
-
-          if(this._subscriptions[hash].onLatest)
-            this._subscriptions[hash].onLatest(hash, message);
-        }
       });
     });
   }
@@ -41,9 +30,9 @@ class Pubsub {
       this._socket.disconnect();
   }
 
-  subscribe(hash, password, callback, onLatest) {
+  subscribe(hash, password, callback) {
     if(!this._subscriptions[hash]) {
-      this._subscriptions[hash] = { head: null, callback: callback, onLatest: onLatest };
+      this._subscriptions[hash] = { head: null, callback: callback };
       this._socket.emit('subscribe', { channel: hash });
     }
   }

src/list/List.js

@@ -12,8 +12,9 @@ class List {
     this._currentBatch = [];
   }
 
+  /* Methods */
   add(data) {
-    const heads = this._findHeads(this.items);
+    const heads = List.findHeads(this.items);
     const node  = new Node(this.id, this.seq, this.ver, data, heads);
     this._currentBatch.push(node);
     this.ver ++;
@@ -36,20 +37,16 @@ class List {
     this.ver = 0;
   }
 
-  _findHeads(list) {
+  /* Private methods */
-    return Lazy(list)
+  _commit() {
-      .reverse()
+    const current = Lazy(this._currentBatch).difference(this._items).toArray();
-      .indexBy((f) => f.id)
+    this._items   = this._items.concat(current);
-      .pairs()
+    this._currentBatch = [];
-      .map((f) => f[1])
+    this.ver = 0;
-      .filter((f) => !this._isReferencedInChain(list, f))
+    this.seq ++;
-      .toArray();
-  }
-
-  _isReferencedInChain(all, item) {
-    return Lazy(all).find((e) => e.hasChild(item)) !== undefined;
   }
 
+  /* Properties */
   get items() {
     return this._items.concat(this._currentBatch);
   }
@@ -67,6 +64,7 @@ class List {
     }
   }
 
+  /* Static methods */
   static fromJson(json) {
     let list = new List(json.id);
     list.seq = json.seq;
@@ -77,6 +75,20 @@ class List {
       .toArray();
     return list;
   }
+
+  static findHeads(list) {
+    return Lazy(list)
+      .reverse()
+      .indexBy((f) => f.id)
+      .pairs()
+      .map((f) => f[1])
+      .filter((f) => !List.isReferencedInChain(list, f))
+      .toArray();
+  }
+
+  static isReferencedInChain(all, item) {
+    return Lazy(all).find((e) => e.hasChild(item)) !== undefined;
+  }
 }
 
 module.exports = List;

src/list/OrbitList.js

@@ -23,52 +23,19 @@ class OrbitList extends List {
     if(this.ver >= MaxBatchSize)
       this._commit();
 
-    const heads = super._findHeads(this.items);
+    const heads = List.findHeads(this.items);
     const node  = new Node(this._ipfs, this.id, this.seq, this.ver, data, heads);
-    node._commit(); // TODO: obsolete?
     this._currentBatch.push(node);
     this.ver ++;
   }
 
   join(other) {
     super.join(other);
-
+    this._fetchHistory(other.items);
-    // WIP: fetch history
-    const isReferenced = (all, item) => _.findLast(all, (f) => f === item) !== undefined;
-    const fetchRecursive = (hash, amount, all, res) => {
-      let result = res ? res : [];
-      hash = hash instanceof Node === true ? hash.hash : hash;
-
-      if(res.length >= amount)
-        return res;
-
-      if(!isReferenced(all, hash)) {
-        all.push(hash);
-        const item = Node.fromIpfsHash(this._ipfs, hash);
-        res.push(item);
-        item.heads.map((head) => fetchRecursive(head, amount, all, res));
-      }
-
-      return res;
-    };
-
-    let allHashes = this._items.map((a) => a.hash);
-
-    const res = _.flatten(other.items.map((e) => _.flatten(e.heads.map((f) => {
-      const remaining = (MaxHistory);
-      return _.flatten(fetchRecursive(f, MaxHistory, allHashes, []));
-    }))));
-
-    res.slice(0, MaxHistory).forEach((item) => {
-      const indices = item.heads.map((k) => _.findIndex(this._items, (b) => b.hash === k));
-      const idx = indices.length > 0 ? Math.max(_.max(indices) + 1, 0) : 0;
-      this._items.splice(idx, 0, item)
-    });
-    // console.log("--> Fetched", res.length, "items from the history\n");
   }
 
   // The LWW-set query interface
-  findAll(opts) {
+  find(opts) {
     let list     = Lazy(this.items);
     const hash   = (opts.gt ? opts.gt : (opts.gte ? opts.gte : (opts.lt ? opts.lt : opts.lte)));
     const amount = opts.amount ? (opts.amount && opts.amount > -1 ? opts.amount : this.items.length) : 1;
@@ -105,14 +72,45 @@ class OrbitList extends List {
     return _findFrom(list.reverse(), hash, amount, opts.lte || !opts.lt).reverse().toArray();
   }
 
-  _commit() {
+  /* Private methods */
-    const current = Lazy(this._currentBatch).difference(this._items).toArray();
+  _fetchHistory(items) {
-    this._items   = this._items.concat(current);
+    let allHashes = this._items.map((a) => a.hash);
-    this._currentBatch = [];
+    const res = Lazy(items)
-    this.ver = 0;
+      .reverse() // Start from the latest item
-    this.seq ++;
+      .map((f) => f.heads).flatten() // Go through all heads
+      .filter((f) => !(f instanceof Node === true)) // OrbitNode vs. {}, filter out instances (we already have them in mem)
+      .map((f) => this._fetchRecursive(f, MaxHistory, allHashes)).flatten() // IO - get the data from IPFS
+      .map((f) => this._insert(f)) // Insert to the list
+      .take(MaxHistory) // How many items from the history we should fetch
+      .toArray();
+    // console.log("--> Fetched", res.length, "items from the history\n");
   }
 
+  _fetchRecursive(hash, amount, all) {
+    const isReferenced = (list, item) => Lazy(list).find((f) => f === item) !== undefined;
+    let result = [];
+    if(!isReferenced(all, hash)) {
+      all.push(hash);
+      const item = await(Node.fromIpfsHash(this._ipfs, hash)); // IO - get from IPFS
+      result.push(item);
+      result = result.concat(Lazy(item.heads)
+        .map((f) => this._fetchRecursive(f, amount, all))
+        .flatten()
+        .toArray());
+    }
+    return result;
+  }
+
+  // Insert to the list right after the latest parent
+  _insert(item) {
+    const index = Lazy(item.heads)
+      .map((next) => Lazy(this._items).map((f) => f.hash).indexOf(next)) // Find the item's parent's indices
+      .reduce((max, a) => a > max ? a : max, 0); // find the largest index (latest parent)
+
+    this._items.splice(index, 0, item);
+  }
+
+  /* Properties */
   get ipfsHash() {
     const toIpfs = async(() => {
       return new Promise(async((resolve, reject) => {
@@ -121,7 +119,8 @@ class OrbitList extends List {
         resolve(list.Hash);
       }));
     });
-    return await(toIpfs());
+    this.hash = await(toIpfs());
+    return this.hash;
   }
 
   get asJson() {

src/list/OrbitNode.js

@@ -64,17 +64,6 @@ class OrbitNode extends Node {
     });
     return await(createNode());
   }
-
-  static hasChild(a, b) {
-    for(let i = 0; i < a.next.length; i ++) {
-      if(typeof a.next[i] instanceof OrbitNode && b.compactId === a.next[i].compactId)
-        return true;
-
-      if(b.compactId === a.next[i])
-        return true;
-    }
-    return false;
-  }
 }
 
 module.exports = OrbitNode;

test/orbit-list-tests.js

@@ -76,8 +76,7 @@ describe('OrbitList', async(function() {
       hash = list.ipfsHash;
       assert.equal(hash, 'Qmecju6aNyQF8LHUNbUrujMmXPfUit7tDkqnmLKLF22aRk');
 
-      const l = await(ipfsAPI.getObject(ipfs, hash));
+      const list2 = List.fromIpfsHash(ipfs, hash);
-      const list2 = List.fromJson(ipfs, JSON.parse(l.Data));
       assert.equal(list2.items[0].data, text1);
       assert.equal(list2.items[1].data, text2);
 
@@ -608,6 +607,75 @@ describe('OrbitList', async(function() {
       assert.equal(list1.items[2].ver, 2);
       done();
     }));
+
+    it('fetches items from history', async((done) => {
+      const list1 = new List(ipfs, 'A');
+      const list2 = new List(ipfs, 'AAA');
+
+      const count = 10;
+      for(let i = 1; i < count + 1; i ++) {
+        list1.add("first " + i);
+        list2.add("second " + i);
+      }
+
+      const hash1 = list1.ipfsHash;
+      const hash2 = list2.ipfsHash;
+      assert.equal(hash1, 'QmaoGci9eiSYdANo63JAkvUpnyXe2uQH1BkwAiKsJHNUWp');
+      assert.equal(hash2, 'QmTXu5g5BzZW3vMBKaXnerZTGXf5XPRFB6y3DXNEmcWWtU');
+
+      const final = new List(ipfs, 'B');
+      const other1 = List.fromIpfsHash(ipfs, hash1);
+      const other2 = List.fromIpfsHash(ipfs, hash2);
+      final.join(other1);
+
+      assert.equal(final.items.length, count);
+      assert.equal(final.items[0].data, "first 1");
+      assert.equal(final.items[final.items.length - 1].data, "first 10");
+
+      final.join(other2);
+      assert.equal(final.items.length, count * 2);
+      assert.equal(final.items[0].data, "first 1");
+      assert.equal(final.items[final.items.length - 1].data, "second 10");
+      done();
+    }));
+
+    it('orders fetched items correctly', async((done) => {
+      const list1 = new List(ipfs, 'A');
+      const list2 = new List(ipfs, 'AAA');
+
+      const count = List.batchSize * 3;
+      for(let i = 1; i < (count * 2) + 1; i ++)
+        list1.add("first " + i);
+
+      const hash1 = list1.ipfsHash;
+      assert.equal(hash1, 'QmaJ2a1AxPBhKcis1HLRnc1UNixSmwd9XBNJzxdnqQSyYa');
+
+      const final = new List(ipfs, 'B');
+      const other1 = List.fromIpfsHash(ipfs, hash1);
+      final.join(other1);
+
+      assert.equal(final.items[0].data, "first 1");
+      assert.equal(final.items[final.items.length - 1].data, "first " + count * 2);
+      assert.equal(final.items.length, count * 2);
+
+      // Second batch
+      for(let i = 1; i < count + 1; i ++)
+        list2.add("second " + i);
+
+      const hash2 = list2.ipfsHash;
+      assert.equal(hash2, 'QmVQ55crzwWY21D7LwMLrxT7aKvCoSVtpo23WRdajSHtBN');
+
+      const other2 = List.fromIpfsHash(ipfs, hash2);
+      final.join(other2);
+
+      // console.log(final.items.map((e) => e.comptactId))
+      assert.equal(final.items.length, count + count * 2);
+      assert.equal(final.items[0].data, "second 1");
+      assert.equal(final.items[1].data, "second 2");
+      assert.equal(final.items[final.items.length - 1].data, "second " + count);
+      done();
+    }));
+
   });
 
   describe('_findHeads', () => {