orbitdb/orbit-db
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

WIP

Browse Source
This commit is contained in:
haad 2016-04-25 17:15:21 +02:00
parent 2d0dda1571
commit 11fd73ab18
14 changed files with 719 additions and 450 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -1,9 +1,8 @@
*sublime*
node_modules/
debug.log
*.log
.vagrant/
.idea/
isolate*.log
dump.rdb
Vagrantfile
orbit-db-cache.json

2
examples/benchmark.js
View File

@ -31,7 +31,7 @@ let run = (async(() => {
// Connect
const ipfs = await(startIpfs());
const orbit = await(OrbitDB.connect(host, port, username, password, ipfs));
const db = await(orbit.channel(channelName));
const db = await(orbit.eventlog(channelName));
// Metrics
let totalQueries = 0;

67
src/Client.js
View File

@ -6,6 +6,7 @@ const PubSub = require('./PubSub');
const OrbitDB = require('./OrbitDB');
const CounterDB = require('./db/CounterDB');
const KeyValueDB = require('./db/KeyValueDB');
const EventLogDB = require('./db/EventLogDB');
class Client {
constructor(ipfs, options) {
@ -15,7 +16,7 @@ class Client {
this.network = null;
this.events = new EventEmitter();
this.options = options || {};
this.db = new OrbitDB(this._ipfs, this.options);
this.eventlogDB = new EventLogDB(this._ipfs, this.options);
this.counterDB = new CounterDB(this._ipfs, this.options);
this.keyvalueDB = new KeyValueDB(this._ipfs, this.options);
}
@ -25,7 +26,7 @@ class Client {
const api = {
put: (key, value) => db.put(dbname, key, value),
set: (key, value) => db.set(dbname, key, value), // alias for put()
get: (key) => db.query(dbname, { key: key }),
get: (key) => db.get(dbname, key),
del: (key) => db.del(dbname, key),
delete: () => db.delete(dbname),
close: () => this._pubsub.unsubscribe(dbname)
@ -47,37 +48,17 @@ class Client {
return this._subscribe(db, dbname, subscribe).then(() => api);
}
channel(channel, password, subscribe) {
if(password === undefined) password = '';
if(subscribe === undefined) subscribe = true;
eventlog(dbname, subscribe) {
const db = this.eventlogDB;
const api = {
iterator: (options) => this._iterator(channel, password, options),
delete: () => this.db.deleteChannel(channel, password),
del: (key) => this.db.del(channel, password, key),
add: (data) => this.db.add(channel, password, data),
put: (key, value) => this.db.put(channel, password, key, value),
get: (key) => {
const items = this._iterator(channel, password, { key: key }).collect();
return items[0] ? items[0] : null;
},
close: () => this._pubsub.unsubscribe(channel)
iterator: (options) => db.iterator(dbname, options),
add: (data) => db.add(dbname, data),
del: (hash) => db.remove(dbname, hash),
delete: () => db.delete(dbname),
close: () => this._pubsub.unsubscribe(dbname)
}
return new Promise((resolve, reject) => {
// Hook to the events from the db and pubsub
this.db.use(channel, this.user).then(() => {
this.db.events[channel].on('write', this._onWrite.bind(this));
this.db.events[channel].on('sync', this._onSync.bind(this));
this.db.events[channel].on('load', this._onLoad.bind(this));
this.db.events[channel].on('loaded', this._onLoaded.bind(this));
if(subscribe)
this._pubsub.subscribe(channel, password, this._onMessage.bind(this));
resolve(api);
}).catch(reject);
});
return this._subscribe(db, dbname, subscribe).then(() => api);
}
disconnect() {
@ -104,8 +85,9 @@ class Client {
}
_onMessage(channel, message) {
this.db.sync(channel, message);
this.counterDB.sync(channel, message);
[this.eventlogDB, this.counterDB, this.keyvalueDB].forEach((db) => db.sync(channel, message))
// this.db.sync(channel, message);
// this.counterDB.sync(channel, message);
}
_onWrite(channel, hash) {
@ -125,27 +107,6 @@ class Client {
this.events.emit('loaded', channel, hash);
}
_iterator(channel, password, options) {
const messages = this.db.query(channel, password, options);
let currentIndex = 0;
let iterator = {
[Symbol.iterator]() {
return this;
},
next() {
let item = { value: null, done: true };
if(currentIndex < messages.length) {
item = { value: messages[currentIndex], done: false };
currentIndex ++;
}
return item;
},
collect: () => messages
}
return iterator;
}
_connect(host, port, username, password, allowOffline) {
return new Promise((resolve, reject) => {
if(allowOffline === undefined) allowOffline = false;

28
src/db/CounterDB.js
View File

@ -11,34 +11,32 @@ class CounterDB extends OrbitDB {
this._counters = {};
}
use(channel, user) {
this._counters[channel] = new Counter(user.username);
return super.use(channel, user);
use(dbname, user) {
this._counters[dbname] = new Counter(user.username);
return super.use(dbname, user);
}
sync(channel, hash) {
// console.log("--> Head:", hash, this.user.username)
super.sync(channel, hash);
const counter = this._counters[channel];
const oplog = this._oplogs[channel];
return oplog.sync(hash)
.then(() => {
sync(dbname, hash) {
const counter = this._counters[dbname];
if(counter) {
return super.sync(dbname, hash).then((oplog) => {
return Lazy(oplog.ops)
.map((f) => Counter.from(f.value))
.map((f) => counter.merge(f))
.toArray();
});
}
}
query(channel) {
return this._counters[channel].value;
query(dbname) {
return this._counters[dbname].value;
}
inc(channel, amount) {
const counter = this._counters[channel];
inc(dbname, amount) {
const counter = this._counters[dbname];
if(counter) {
counter.increment(amount);
return this._write(channel, '', OpTypes.Inc, null, counter.payload);
return this._write(dbname, '', OpTypes.Inc, null, counter.payload);
}
}
}

111
src/db/EventLogDB.js Normal file
View File

@ -0,0 +1,111 @@
'use strict';
const Lazy = require('lazy.js');
const OrbitDB = require('./OrbitDB');
const OpTypes = require('./Operation').Types;
const GSet = require('./GSet');
class EventLogDB extends OrbitDB {
constructor(ipfs, options) {
super(ipfs, options)
this._set = null;
// this._counters = {};
}
use(name, user) {
this._set = new GSet(user.username);
return super.use(name, user);
}
sync(dbname, hash) {
return super.sync(dbname, hash).then((oplog) => {
return Lazy(oplog.ops)
.map((f) => GSet.from(f.value))
.map((f) => this._set.merge(f))
.toArray();
});
}
add(dbname, data) {
const oplog = this._oplogs[dbname];
if(oplog) {
return oplog.addOperation(dbname, OpTypes.Add, null, data).then((result) => {
this.events[dbname].emit('write', dbname, result.hash);
this._set.add(result.op.hash);
// console.log("OP", result)
return result.op.hash;
});
}
return;
}
remove(dbname, hash) {
const oplog = this._oplogs[dbname];
if(oplog) {
return oplog.addOperation(dbname, OpTypes.Delete, hash).then((result) => {
this.events[dbname].emit('write', dbname, result.hash);
this._set.remove(hash);
// console.log("OP", result)
return result.op.hash;
});
}
return;
}
iterator(dbname, options) {
const messages = this.query(dbname, options);
let currentIndex = 0;
let iterator = {
[Symbol.iterator]() {
return this;
},
next() {
let item = { value: null, done: true };
if(currentIndex < messages.length) {
item = { value: messages[currentIndex], done: false };
currentIndex ++;
}
return item;
},
collect: () => messages
}
return iterator;
}
query(dbname, opts) {
if(!opts) opts = {};
const oplog = this._oplogs[dbname];
const amount = opts.limit ? (opts.limit > -1 ? opts.limit : oplog.ops.length) : 1; // Return 1 if no limit is provided
let result = [];
if(opts.gt || opts.gte) {
// Greater than case
console.log("2")
result = this._read(this._set.value, opts.gt ? opts.gt : opts.gte, amount, opts.gte ? opts.gte : false)
} else {
// Lower than and lastN case, search latest first by reversing the sequence
result = this._read(this._set.value.reverse(), opts.lt ? opts.lt : opts.lte, amount, opts.lte || !opts.lt).reverse()
}
if(opts.reverse) result.reverse();
let res = result.toArray();
// const removed = this._itemsR.find((e) => e === item);
res = oplog.ops.filter((f) => res.find((e) => e === f.hash))
// console.log("RSULT", res)
return res;
}
_read(ops, key, amount, inclusive) {
// console.log("KET", key, amount, inclusive)
return Lazy(ops)
.skipWhile((f) => key && f !== key) // Drop elements until we have the first one requested
.drop(inclusive ? 0 : 1) // Drop the 'gt/lt' item, include 'gte/lte' item
.take(amount);
}
}
module.exports = EventLogDB;

57
src/db/GSet.js Normal file
View File

@ -0,0 +1,57 @@
'use strict';
const isEqual = require('./utils').isEqual;
class GSet {
constructor(id, payload) {
this.id = id;
this._added = {};
this._removed = {};
}
add(data, ts) {
this._added[data] = { ts: ts || new Date().getTime() }
}
remove(data, ts) {
this._removed[data] = { ts: ts || new Date().getTime() }
}
get value() {
// console.log("AAA", this._added, this._removed)
return Object.keys(this._added).map((f) => {
const removed = this._removed[f];
// console.log("--", removed, this._added[f]);
if(!removed || (removed && removed.ts < this._added[f].ts)) {
return f;
}
return null;
}).filter((f) => f !== null)
.map((f) => {
console.log("f", f)
return f;
});
}
compare(other) {
return false;
// if(other.id !== this.id)
// return false;
// return isEqual(other._counters, this._counters);
}
merge(other) {
// Object.keys(other._counters).forEach((f) => {
// this._counters[f] = Math.max(this._counters[f] ? this._counters[f] : 0, other._counters[f]);
// });
}
static from(payload) {
return new GSet(payload.id, payload.items);
}
}
module.exports = GSet;

95
src/db/KeyValueDB.js
View File

@ -3,62 +3,77 @@
const Lazy = require('lazy.js');
const OrbitDB = require('./OrbitDB');
const OpTypes = require('./Operation').Types;
const Counter = require('./GCounter');
const GSet = require('./GSet');
class KeyValueDB extends OrbitDB {
constructor(ipfs, options) {
super(ipfs, options)
// this._counters = {};
// this._set = null;
}
use(name, user) {
// this._counters[name] = new Counter(user.username);
// this._set = new GSet(user.username);
return super.use(name, user);
}
sync(name, hash) {
// console.log("--> Head:", hash, this.user.username)
super.sync(name, hash);
// const counter = this._counters[name];
const oplog = this._oplogs[name];
return oplog.sync(hash)
.then(() => {
return Lazy(oplog.ops)
// .map((f) => Counter.from(f.value))
// .map((f) => counter.merge(f))
.toArray();
sync(dbname, hash) {
return super.sync(dbname, hash).then((oplog) => {
return Lazy(oplog.ops)
// .map((f) => GSet.from(f.value))
// .map((f) => this._set.merge(f))
.toArray();
});
}
put(dbname, key, data) {
// set.add(data);
const oplog = this._oplogs[dbname];
if(oplog) {
return oplog.addOperation(dbname, OpTypes.Put, key, data).then((result) => {
this.events[dbname].emit('write', dbname, result.hash);
// console.log("OP", result);
// this._set.add(result.op.hash, result.op.meta.ts);
return result.op.hash;
});
}
// return this._write(dbname, '', OpTypes.Put, key, data).then((op) => {
// console.log("OP", op);
// // this._set.add(op);
// })
}
put(name, key, data) {
return this._write(name, '', OpTypes.Put, key, data);
set(dbname, key, data) {
this.put(dbname, key, data);
}
set(name, key, data) {
this.put(name, key, data);
del(dbname, key) {
const oplog = this._oplogs[dbname];
if(oplog) {
return oplog.addOperation(dbname, OpTypes.Delete, key).then((result) => {
// console.log("OP", op);
return result.op.hash;
});
}
}
del(name, key) {
return this._write(name, '', OpTypes.Delete, key);
get(dbname, key) {
if(!key)
return;
const oplog = this._oplogs[dbname];
// console.log("INIT", JSON.stringify(this._set.value, null, 2), oplog.ops)
const items = oplog.ops.filter((f) => f.key === key)
console.log("ITEM", items, key)
let result = this._read(oplog.ops.reverse(), key, 1, true).toArray()[0];
// result = this._read(operations.reverse(), opts.key, 1, true).map((f) => f.value);
// let result = this._read(this._set.value, key).toArray()[0];
// let result = this._read(this._set.value, key).toArray()[0];
console.log("RSULT", result)
// result = oplog.ops.find((e) => e.hash === result).value;
return result ? result.value : null;
}
query(name, opts) {
this.events[name].emit('load', 'query', name);
if(!opts) opts = {};
let result = [];
const oplog = this._oplogs[name];
// Key-Value, search latest key first
if(opts.key)
result = this._read(oplog, opts.key).map((f) => f.value).toArray();
this.events[name].emit('loaded', 'query', name);
return result.length > 0 ? result[0] : null;
}
_read(oplog, key) {
// Last-Write-Wins, ie. use only the first occurance of the key
_read(ops, key) {
let handled = [];
const _createLWWSet = (item) => {
if(Lazy(handled).indexOf(item.key) === -1) {
@ -70,12 +85,14 @@ class KeyValueDB extends OrbitDB {
};
// Find the items from the sequence (list of operations)
return Lazy(oplog.ops.reverse())
.compact()
return Lazy(ops)
.skipWhile((f) => key && f.key !== key) // Drop elements until we have the first one requested
.map(_createLWWSet) // Return items as LWW (ignore values after the first found)
.compact() // Remove nulls
.take(1);
// return Lazy(ops)
// .skipWhile((f) => key && f !== key) // Drop elements until we have the first one requested
// .take(1);
}
}

57
src/db/LWWSet.js Normal file
View File

@ -0,0 +1,57 @@
'use strict';
const isEqual = require('./utils').isEqual;
class GSet {
constructor(id, payload) {
this.id = id;
this._items = payload ? payload : [];
// this._counters[this.id] = this._counters[this.id] ? this._counters[this.id] : 0;
}
add(data) {
this._items.push(data);
}
// remove(data) {
// }
// increment(amount) {
// if(!amount) amount = 1;
// this._counters[this.id] = this._counters[this.id] + amount;
// }
// get value() {
// return Object.keys(this._counters)
// .map((f) => this._counters[f])
// .reduce((previousValue, currentValue) => previousValue + currentValue, 0);
// }
query() {
return this._items;
}
get payload() {
return { id: this.id, items: this._items };
}
// compare(other) {
// if(other.id !== this.id)
// return false;
// return isEqual(other._counters, this._counters);
// }
// merge(other) {
// Object.keys(other._counters).forEach((f) => {
// this._counters[f] = Math.max(this._counters[f] ? this._counters[f] : 0, other._counters[f]);
// });
// }
// static from(payload) {
// return new LWWSet(payload.id, payload.counters);
// }
}
module.exports = GSet;

18
src/db/Operation.js
View File

@ -6,16 +6,16 @@ const Post = require('../post/Post');
class Operation {
static create(ipfs, log, user, operation, key, value, data) {
return new Promise((resolve, reject) => {
// return new Promise((resolve, reject) => {
let post;
Operation._createOperation(ipfs, user, operation, key, value)
.then((op) => {
post = op.Post;
return log.add(op.Hash);
})
.then((node) => resolve({ node: node, op: post }))
.catch(reject);
});
return Operation._createOperation(ipfs, user, operation, key, value)
// .then((op) => {
// post = op.Post;
// return log.add(op.Hash);
// })
// .then((node) => resolve({ node: node, op: post }))
// .catch(reject);
// });
}
static _createOperation(ipfs, user, operation, key, value) {

69
src/db/OperationsLog.js
View File

@ -1,14 +1,19 @@
'use strict';
const EventEmitter = require('events').EventEmitter;
const Lazy = require('lazy.js');
const Log = require('ipfs-log');
const Lazy = require('lazy.js');
const Log = require('ipfs-log');
const Cache = require('../Cache');
const DBOperation = require('./Operation');
/*
Load, cache and index operations log
*/
class OperationsLog {
constructor(ipfs, options) {
constructor(ipfs, dbname, opts) {
this.dbname = dbname;
this.options = opts || { cacheFile: null };
this.id = null;
this.options = options;
this.events = new EventEmitter();
this.lastWrite = null;
this._ipfs = ipfs;
this._log = null;
@ -21,7 +26,20 @@ class OperationsLog {
create(user) {
this.id = user.username;
return Log.create(this._ipfs, this.id).then((log) => this._log = log);
return Log.create(this._ipfs, this.id)
.then((log) => this._log = log)
.then(() => {
if(this.options.cacheFile)
return Cache.loadCache(this.options.cacheFile)
return;
})
.then(() => {
if(this.options.cacheFile)
return this.sync(Cache.get(this.dbname))
return;
});
}
delete() {
@ -33,7 +51,6 @@ class OperationsLog {
if(!hash || hash === this.lastWrite || !this._log)
return Promise.resolve();
this.events.emit('load');
const oldCount = this._log.items.length;
return Log.fromIpfsHash(this._ipfs, hash)
@ -43,10 +60,38 @@ class OperationsLog {
return;
return this._cacheInMemory(this._log);
}).then(() => {
this.events.emit('sync');
this.events.emit('loaded');
return;
})
.then(() => Cache.set(this.id, hash));
}
addOperation(dbname, operation, key, value) {
let post;
return DBOperation.create(this._ipfs, this._log, this.user, operation, key, value)
// .then((op) => {
// post = op.Post;
// return log.add(op.Hash);
// })
// .then((node) => resolve({ node: node, op: post }))
.then((result) => {
// console.log("res1", result)
return this._log.add(result.Hash).then((node) => {
return { node: node, op: result.Post };
});
})
.then((result) => {
// console.log("res2", result)
this._cachePayload(result.node.payload, result.op);
return result;
}).then((result) => {
return Log.getIpfsHash(this._ipfs, this._log)
.then((listHash) => {
this.lastWrite = listHash;
Cache.set(this.dbname, listHash);
// this.events[dbname].emit('write', this.dbname, listHash);
return { hash: listHash, op: result.op };
});
}).then((result) => {
return result;
});
}

57
src/db/OrbitDB.js
View File

@ -3,8 +3,6 @@
const Lazy = require('lazy.js');
const EventEmitter = require('events').EventEmitter;
const Log = require('ipfs-log');
const Cache = require('../Cache');
const DBOperation = require('./Operation');
const OperationsLog = require('./OperationsLog');
class OrbitDB {
@ -15,46 +13,53 @@ class OrbitDB {
this._oplogs = {};
}
use(channel, user) {
use(dbname, user) {
this.user = user;
this.events[channel] = new EventEmitter();
this._oplogs[channel] = new OperationsLog(this._ipfs, this.options);
return this._oplogs[channel].create(user)
.then(() => {
if(this.options.cacheFile)
return Cache.loadCache(this.options.cacheFile)
})
.then(() => {
if(this.options.cacheFile)
return this.sync(channel, Cache.get(channel))
});
this.events[dbname] = new EventEmitter();
this._oplogs[dbname] = new OperationsLog(this._ipfs, dbname);
this.events[dbname].emit('load');
return this._oplogs[dbname].create(user)
}
sync(channel, hash) {
Cache.set(channel, hash);
sync(dbname, hash) {
// console.log("--> Head:", hash)
const oplog = this._oplogs[dbname];
if(oplog) {
this.events[dbname].emit('load');
return oplog.sync(hash)
.then(() => this.events[dbname].emit('sync'))
.then(() => oplog);
}
return Promise.resolve();
}
query(channel) {
query(dbname) {
}
delete(channel) {
if(this._oplogs[channel])
this._oplogs[channel].delete();
delete(dbname) {
if(this._oplogs[dbname])
this._oplogs[dbname].delete();
}
_write(channel, password, operation, key, value) {
const oplog = this._oplogs[channel];
_write(dbname, password, operation, key, value) {
const oplog = this._oplogs[dbname];
const log = oplog._log;
return DBOperation.create(this._ipfs, log, this.user, operation, key, value)
.then((result) => {
oplog._cachePayload(result.node.payload, result.op);
return result;
// console.log("res", result)
return log.add(result.Hash);
})
.then((result) => {
// console.log("res", result)
oplog._cachePayload(result.node.payload, result.op);
return result;
}).then((result) => {
return Log.getIpfsHash(this._ipfs, log)
.then((listHash) => {
oplog.lastWrite = listHash;
Cache.set(channel, listHash);
this.events[channel].emit('write', channel, listHash);
Cache.set(dbname, listHash);
this.events[dbname].emit('write', dbname, listHash);
return result;
});
}).then((result) => result.node.payload);

63
test/client.test.js
View File

@ -14,12 +14,20 @@ require('logplease').setLogLevel('ERROR');
// Orbit
const username = 'testrunner';
const password = '';
const ipfsPath = '/tmp/orbittests';
const startIpfs = () => {
return new Promise((resolve, reject) => {
ipfsd.disposableApi((err, ipfs) => {
if(err) console.error(err);
resolve(ipfs);
// ipfsd.disposableApi((err, ipfs) => {
// if(err) console.error(err);
// resolve(ipfs);
// });
ipfsd.local((err, node) => {
if(err) reject(err);
node.startDaemon((err, ipfs) => {
if(err) reject(err);
resolve(ipfs);
});
});
});
};
@ -37,8 +45,8 @@ describe('Orbit Client', function() {
try {
ipfs = await(startIpfs());
client = await(OrbitClient.connect('localhost', 3333, username, password, ipfs, { allowOffline: true }));
db = await(client.channel(channel, '', false));
db.delete();
// db = await(client.channel(channel, '', false));
// db.delete();
} catch(e) {
console.log(e);
assert.equal(e, null);
@ -52,6 +60,7 @@ describe('Orbit Client', function() {
if(client) client.disconnect();
});
/*
describe('API', function() {
let api;
@ -122,11 +131,13 @@ describe('Orbit Client', function() {
done();
}));
});
*/
describe('Add events', function() {
beforeEach(() => {
beforeEach(async(() => {
db = await(client.eventlog(channel, false));
db.delete();
});
}));
it('adds an item to an empty channel', async((done) => {
const head = await(db.add('hello'));
@ -147,12 +158,13 @@ describe('Orbit Client', function() {
}));
it('adds five items', async((done) => {
for(let i = 0; i < 5; i ++) {
let hash = await(db.add('hello'));
assert.notEqual(hash, null);
assert.equal(hash.startsWith('Qm'), true);
assert.equal(hash.length, 46);
}
for(let i = 1; i <= 5; i ++)
await(db.add('hello' + i));
const items = db.iterator({ limit: -1 }).collect();
assert.equal(items.length, 5);
assert.equal(_.first(items.map((f) => f.value)), 'hello1');
assert.equal(_.last(items.map((f) => f.value)), 'hello5');
done();
}));
@ -168,18 +180,18 @@ describe('Orbit Client', function() {
});
describe('Delete events', function() {
beforeEach(() => {
beforeEach(async(() => {
db = await(client.eventlog(channel, false));
db.delete();
// const items = db.iterator().collect();
// assert.equal(items.length, 0);
});
}));
it('deletes an item when only one item in the database', async((done) => {
const head = await(db.add('hello1'));
let item = db.iterator().collect();
const delop = await(db.del(head));
const items = db.iterator().collect();
console.log(items);
assert.equal(delop.startsWith('Qm'), true);
assert.equal(items.length, 0);
done();
@ -189,7 +201,7 @@ describe('Orbit Client', function() {
await(db.add('hello1'));
const head = await(db.add('hello2'));
await(db.del(head));
const items = db.iterator().collect();
const items = db.iterator({ limit: -1 }).collect();
assert.equal(items.length, 1);
assert.equal(items[0].value, 'hello1');
done();
@ -215,6 +227,7 @@ describe('Orbit Client', function() {
beforeEach(async((done) => {
items = [];
db = await(client.eventlog(channel, false));
db.delete();
for(let i = 0; i < itemCount; i ++) {
const hash = await(db.add('hello' + i));
@ -511,23 +524,29 @@ describe('Orbit Client', function() {
describe('Delete', function() {
it('deletes a channel from the local database', () => {
const result = db.delete();
assert.equal(result, true);
// assert.equal(result, true);
const iter = db.iterator();
assert.equal(iter.next().value, null);
});
});
describe('Key-Value Store', function() {
before(() => {
// before(() => {
// db.delete();
// });
beforeEach(async((done) => {
db = await(client.kvstore(channel, '', false));
db.delete();
});
done();
}));
afterEach(() => {
db.delete();
});
it('put', async((done) => {
db = await(client.kvstore(channel, '', false));
// db = await(client.kvstore(channel, '', false));
await(db.put('key1', 'hello!'));
const value = db.get('key1');
// let all = db.iterator().collect();

206
test/counterdb.test.js
View File

@ -1,115 +1,115 @@
'use strict';
// 'use strict';
const assert = require('assert');
const Promise = require('bluebird');
const rimraf = require('rimraf')
const ipfsd = require('ipfsd-ctl');
const OrbitClient = require('../src/Client');
const OrbitServer = require('orbit-server/src/server');
// const assert = require('assert');
// const Promise = require('bluebird');
// const rimraf = require('rimraf')
// const ipfsd = require('ipfsd-ctl');
// const OrbitClient = require('../src/Client');
// const OrbitServer = require('orbit-server/src/server');
// Mute logging
require('logplease').setLogLevel('ERROR');
// // Mute logging
// require('logplease').setLogLevel('ERROR');
const username = 'testrunner';
const username2 = 'rennurtset';
// const username = 'testrunner';
// const username2 = 'rennurtset';
const ipfsPath = '/tmp/orbittests';
// const ipfsPath = '/tmp/orbittests';
const startIpfs = () => {
return new Promise((resolve, reject) => {
// ipfsd.local(ipfsPath, (err, node) => {
// if(err) reject(err);
// node.startDaemon((err, ipfs) => {
// if(err) reject(err);
// resolve(ipfs);
// });
// });
OrbitServer.start();
ipfsd.disposableApi((err, ipfs) => {
if(err) reject(err);
resolve(ipfs);
});
});
};
// const startIpfs = () => {
// return new Promise((resolve, reject) => {
// // ipfsd.local(ipfsPath, (err, node) => {
// // if(err) reject(err);
// // node.startDaemon((err, ipfs) => {
// // if(err) reject(err);
// // resolve(ipfs);
// // });
// // });
// OrbitServer.start();
// ipfsd.disposableApi((err, ipfs) => {
// if(err) reject(err);
// resolve(ipfs);
// });
// });
// };
describe('Orbit Client', function() {
this.timeout(20000);
// describe('Orbit Client', function() {
// this.timeout(20000);
let ipfs, client1, client2;
// let ipfs, client1, client2;
before((done) => {
rimraf.sync('./orbit-db-cache.json')
startIpfs().then((res) => {
ipfs = res;
Promise.map([username, username2], (login) => {
return OrbitClient.connect('localhost', 3333, login, '', ipfs, { allowOffline: false, cacheFile: './orbit-db-cache.json' });
}).then((clients) => {
client1 = clients[0];
client2 = clients[1];
done();
}).catch((e) => {
console.log(e.stack);
assert.equal(e, null);
});
});
});
// before((done) => {
// rimraf.sync('./orbit-db-cache.json')
// startIpfs().then((res) => {
// ipfs = res;
// Promise.map([username, username2], (login) => {
// return OrbitClient.connect('localhost', 3333, login, '', ipfs, { allowOffline: false, cacheFile: './orbit-db-cache.json' });
// }).then((clients) => {
// client1 = clients[0];
// client2 = clients[1];
// done();
// }).catch((e) => {
// console.log(e.stack);
// assert.equal(e, null);
// });
// });
// });
after((done) => {
if(client1) client1.disconnect();
if(client2) client2.disconnect();
rimraf('./orbit-db-cache.json', done)
});
// after((done) => {
// if(client1) client1.disconnect();
// if(client2) client2.disconnect();
// rimraf('./orbit-db-cache.json', done)
// });
describe('counters', function() {
it('increases a counter value', (done) => {
client1.counter('counter test', false).then((counter) => {
Promise.map([13, 1], (f) => counter.inc(f), { concurrency: 1 }).then(() => {
assert.equal(counter.value(), 14);
done();
}).catch((e) => {
console.error(e.stack);
assert.equal(null, e);
done();
});
}).catch((e) => {
console.error(e.stack);
assert.equal(' ', e.message);
done();
});
});
// describe('counters', function() {
// it('increases a counter value', (done) => {
// client1.counter('counter test', false).then((counter) => {
// Promise.map([13, 1], (f) => counter.inc(f), { concurrency: 1 }).then(() => {
// assert.equal(counter.value(), 14);
// done();
// }).catch((e) => {
// console.error(e.stack);
// assert.equal(null, e);
// done();
// });
// }).catch((e) => {
// console.error(e.stack);
// assert.equal(' ', e.message);
// done();
// });
// });
it('creates a new counter from cached data', function(done) {
client1.counter('counter test', false).then((counter) => {
assert.equal(counter.value(), 14);
done();
}).catch((e) => {
console.error(e.stack);
assert.equal(' ', e.message);
done();
});
});
// it('creates a new counter from cached data', function(done) {
// client1.counter('counter test', false).then((counter) => {
// assert.equal(counter.value(), 14);
// done();
// }).catch((e) => {
// console.error(e.stack);
// assert.equal(' ', e.message);
// done();
// });
// });
it('syncs counters', (done) => {
const name = new Date().getTime();
Promise.all([client1.counter(name), client2.counter(name)]).then((counters) => {
const res1 = Promise.map([13, 10], (f) => counters[1].inc(f), { concurrency: 1 });
const res2 = Promise.map([2, 5], (f) => counters[0].inc(f), { concurrency: 1 })
Promise.all([res1, res2]).then((res) => {
setTimeout(() => {
assert.equal(counters[0].value(), 30);
assert.equal(counters[1].value(), 30);
done();
}, 1000)
}).catch((e) => {
console.log(e);
assert(e);
done();
});
}).catch((e) => {
console.log(e);
assert(e);
done();
});
});
});
});
// it('syncs counters', (done) => {
// const name = new Date().getTime();
// Promise.all([client1.counter(name), client2.counter(name)]).then((counters) => {
// const res1 = Promise.map([13, 10], (f) => counters[1].inc(f), { concurrency: 1 });
// const res2 = Promise.map([2, 5], (f) => counters[0].inc(f), { concurrency: 1 })
// Promise.all([res1, res2]).then((res) => {
// setTimeout(() => {
// assert.equal(counters[0].value(), 30);
// assert.equal(counters[1].value(), 30);
// done();
// }, 1000)
// }).catch((e) => {
// console.log(e);
// assert(e);
// done();
// });
// }).catch((e) => {
// console.log(e);
// assert(e);
// done();
// });
// });
// });
// });

336
test/orbit-db.test.js
View File

@ -1,196 +1,196 @@
'use strict';
// 'use strict';
const _ = require('lodash');
const fs = require('fs');
const path = require('path');
const assert = require('assert');
const async = require('asyncawait/async');
const await = require('asyncawait/await');
const ipfsd = require('ipfsd-ctl');
const Log = require('ipfs-log');
const OrbitDB = require('../src/OrbitDB');
// const _ = require('lodash');
// const fs = require('fs');
// const path = require('path');
// const assert = require('assert');
// const async = require('asyncawait/async');
// const await = require('asyncawait/await');
// const ipfsd = require('ipfsd-ctl');
// const Log = require('ipfs-log');
// const OrbitDB = require('../src/OrbitDB');
// Mute logging
require('logplease').setLogLevel('ERROR');
// // Mute logging
// require('logplease').setLogLevel('ERROR');
// Orbit
const username = 'testrunner';
const password = '';
const user = { username: username };
// // Orbit
// const username = 'testrunner';
// const password = '';
// const user = { username: username };
const startIpfs = () => {
return new Promise((resolve, reject) => {
ipfsd.disposableApi((err, ipfs) => {
if(err) console.error(err);
resolve(ipfs);
});
});
};
// const startIpfs = () => {
// return new Promise((resolve, reject) => {
// ipfsd.disposableApi((err, ipfs) => {
// if(err) console.error(err);
// resolve(ipfs);
// });
// });
// };
describe('OrbitDB', function() {
this.timeout(3000);
// describe('OrbitDB', function() {
// this.timeout(3000);
let ipfs, db;
let channel = 'orbit-db.test';
// let ipfs, db;
// let channel = 'orbit-db.test';
before(async(function(done) {
this.timeout(20000);
try {
ipfs = await(startIpfs());
} catch(e) {
console.log(e);
assert.equal(e, null);
}
done();
}));
// before(async(function(done) {
// this.timeout(20000);
// try {
// ipfs = await(startIpfs());
// } catch(e) {
// console.log(e);
// assert.equal(e, null);
// }
// done();
// }));
after(() => {
if(db) db.delete();
});
// after(() => {
// if(db) db.delete();
// });
describe('constructor', function() {
it('sets defaults', async((done) => {
db = new OrbitDB(ipfs);
assert.notEqual(db._ipfs, null);
assert.notEqual(db._logs, null);
assert.notEqual(db.options, null);
assert.equal(db.lastWrite, null);
assert.equal(db._cached.length, 0);
done();
}));
// describe('constructor', function() {
// it('sets defaults', async((done) => {
// db = new OrbitDB(ipfs);
// assert.notEqual(db._ipfs, null);
// assert.notEqual(db._logs, null);
// assert.notEqual(db.options, null);
// assert.equal(db.lastWrite, null);
// assert.equal(db._cached.length, 0);
// done();
// }));
it('sets options', async((done) => {
db = new OrbitDB(ipfs, { option1: 'hello', option2: 2 });
assert.equal(db.options.option1, 'hello');
assert.equal(db.options.option2, 2);
done();
}));
});
// it('sets options', async((done) => {
// db = new OrbitDB(ipfs, { option1: 'hello', option2: 2 });
// assert.equal(db.options.option1, 'hello');
// assert.equal(db.options.option2, 2);
// done();
// }));
// });
describe('use', function() {
beforeEach(() => {
db = new OrbitDB(ipfs);
});
// describe('use', function() {
// beforeEach(() => {
// db = new OrbitDB(ipfs);
// });
it('sets user', (done) => {
db.use(channel, user).then(() => {
assert.equal(db.user.username, username);
done();
});
});
// it('sets user', (done) => {
// db.use(channel, user).then(() => {
// assert.equal(db.user.username, username);
// done();
// });
// });
it('creates an empty log for the channel', (done) => {
db.use(channel, user).then(() => {
assert(db._logs[channel]);
assert.equal(db._logs[channel].id, username);
assert.equal(db._logs[channel].items.length, 0);
done();
});
});
// it('creates an empty log for the channel', (done) => {
// db.use(channel, user).then(() => {
// assert(db._logs[channel]);
// assert.equal(db._logs[channel].id, username);
// assert.equal(db._logs[channel].items.length, 0);
// done();
// });
// });
it('creates event emitter for the channel', (done) => {
const EventEmitter = require('events').EventEmitter;
db.use(channel, user).then(() => {
assert(db.events[channel]);
assert.equal(db.events[channel] instanceof EventEmitter, true);
done();
});
});
});
// it('creates event emitter for the channel', (done) => {
// const EventEmitter = require('events').EventEmitter;
// db.use(channel, user).then(() => {
// assert(db.events[channel]);
// assert.equal(db.events[channel] instanceof EventEmitter, true);
// done();
// });
// });
// });
describe('sync', function() {
let log, otherLogHash, otherDbHash;
// describe('sync', function() {
// let log, otherLogHash, otherDbHash;
beforeEach(async((done) => {
log = await(Log.create(ipfs, username));
await(log.add("one"));
await(log.add("two"));
await(log.add("three"));
otherLogHash = await(Log.getIpfsHash(ipfs, log));
// beforeEach(async((done) => {
// log = await(Log.create(ipfs, username));
// await(log.add("one"));
// await(log.add("two"));
// await(log.add("three"));
// otherLogHash = await(Log.getIpfsHash(ipfs, log));
const cacheFile = path.join(process.cwd(), '/test', 'orbit-db-test-cache.json');
// const cacheFile = path.join(process.cwd(), '/test', 'orbit-db-test-cache.json');
let count = 0;
const db2 = new OrbitDB(ipfs);
await(db2.use(channel, user));
db2.events[channel].on('write', async((channel, hash) => {
otherDbHash = hash;
if(count === 2) {
const obj = Object.defineProperty({}, channel, {
value: hash,
writable: true
});
fs.writeFileSync(cacheFile, JSON.stringify(obj));
// let count = 0;
// const db2 = new OrbitDB(ipfs);
// await(db2.use(channel, user));
// db2.events[channel].on('write', async((channel, hash) => {
// otherDbHash = hash;
// if(count === 2) {
// const obj = Object.defineProperty({}, channel, {
// value: hash,
// writable: true
// });
// fs.writeFileSync(cacheFile, JSON.stringify(obj));
db = new OrbitDB(ipfs, { cacheFile: cacheFile });
await(db.use(channel, user));
done();
} else {
count ++;
}
}));
await(db2.add(channel, '', "hello world 1"));
await(db2.add(channel, '', "hello world 2"));
await(db2.add(channel, '', "hello world 3"));
}));
// db = new OrbitDB(ipfs, { cacheFile: cacheFile });
// await(db.use(channel, user));
// done();
// } else {
// count ++;
// }
// }));
// await(db2.add(channel, '', "hello world 1"));
// await(db2.add(channel, '', "hello world 2"));
// await(db2.add(channel, '', "hello world 3"));
// }));
afterEach(() => {
db = null;
});
// afterEach(() => {
// db = null;
// });
describe('events', function() {
it('emits \'loaded\' event when sync hash is null', async((done) => {
db.events[channel].on('loaded', (src, channelName) => done());
db.sync(channel, null);
}));
// describe('events', function() {
// it('emits \'loaded\' event when sync hash is null', async((done) => {
// db.events[channel].on('loaded', (src, channelName) => done());
// db.sync(channel, null);
// }));
it('emits \'load\' event when sync starts', async((done) => {
db.events[channel].on('load', (src, channelName) => done());
db.sync(channel, otherDbHash);
}));
// it('emits \'load\' event when sync starts', async((done) => {
// db.events[channel].on('load', (src, channelName) => done());
// db.sync(channel, otherDbHash);
// }));
it('emits \'loaded\' event when sync finishes', async((done) => {
db.events[channel].on('loaded', (src, channelName) => done());
db.sync(channel, otherDbHash);
}));
// it('emits \'loaded\' event when sync finishes', async((done) => {
// db.events[channel].on('loaded', (src, channelName) => done());
// db.sync(channel, otherDbHash);
// }));
it('emits \'sync\' event if items were merged', async((done) => {
db.events[channel].on('sync', (channelName, hash) => {
assert.equal(channelName, channel);
assert.equal(hash, otherDbHash);
done();
});
db.sync(channel, otherDbHash);
}));
// it('emits \'sync\' event if items were merged', async((done) => {
// db.events[channel].on('sync', (channelName, hash) => {
// assert.equal(channelName, channel);
// assert.equal(hash, otherDbHash);
// done();
// });
// db.sync(channel, otherDbHash);
// }));
it('doesn\'t emit \'sync\' event if items weren\'t merged', async((done) => {
db._logs[channel] = log;
db.events[channel].on('sync', (channelName, hash) => {
assert.equal(false, true);
done();
});
db.events[channel].on('loaded', (src, channelName) => done());
db.sync(channel, otherLogHash);
}));
});
// it('doesn\'t emit \'sync\' event if items weren\'t merged', async((done) => {
// db._logs[channel] = log;
// db.events[channel].on('sync', (channelName, hash) => {
// assert.equal(false, true);
// done();
// });
// db.events[channel].on('loaded', (src, channelName) => done());
// db.sync(channel, otherLogHash);
// }));
// });
describe('cache payloads', function() {
it('fetches payloads', (done) => {
assert.equal(db._cached.length, 0);
db.events[channel].on('loaded', (src, channelName) => {
assert.equal(db._cached.length, 3);
done();
});
db.sync(channel, otherDbHash);
});
// describe('cache payloads', function() {
// it('fetches payloads', (done) => {
// assert.equal(db._cached.length, 0);
// db.events[channel].on('loaded', (src, channelName) => {
// assert.equal(db._cached.length, 3);
// done();
// });
// db.sync(channel, otherDbHash);
// });
it('throws an error if fetching went wrong', (done) => {
db.sync(channel, otherLogHash).catch((e) => {
assert.equal(e.message, 'invalid ipfs ref path');
done();
})
});
});
// it('throws an error if fetching went wrong', (done) => {
// db.sync(channel, otherLogHash).catch((e) => {
// assert.equal(e.message, 'invalid ipfs ref path');
// done();
// })
// });
// });
});
});
// });
// });