Merge 3305a72 into c8a2325

README.md

@@ -86,7 +86,7 @@ Creates a new FeedStore `without wait for their initialization.`
 
 > The initialization happens by running: `await feedStore.initialize()`
 
-#### `await feedStore.openFeed(path, [options]) -> Hypercore`
+#### `feedStore.openFeed(path, [options]) -> Promise<Hypercore>`
 
 Creates a new hypercore feed identified by a string path.
 
@@ -97,29 +97,29 @@ Creates a new hypercore feed identified by a string path.
   - `metadata: Object`: Serializable object with custom data about the feed.
   - `[...hypercoreOptions]`: Hypercore options.
 
-#### `await feedStore.closeFeed(path)`
+#### `feedStore.closeFeed(path) -> Promise`
 
 Close a feed by the path.
 
-#### `await feedStore.deleteDescriptor(path)`
+#### `feedStore.deleteDescriptor(path) -> Promise`
 
 Remove a descriptor from the database by the path.
 
 > This operation would not close the feed.
 
-#### `await feedStore.close()`
+#### `feedStore.close() -> Promise`
 
 Close the hypertrie database and their feeds.
 
-#### `await feedStore.loadFeeds((descriptor) => Boolean) -> Hypercore[]`
+#### `feedStore.loadFeeds((descriptor) => Boolean) -> Promise<Hypercore[]>`
 
 Loads feeds using a function to filter what feeds you want to load from the database.
 
 ```javascript
 const feeds = await feedStore.loadFeeds(descriptor => descriptor.metadata.tag === 'important')
 ```
 
-#### `await feedStore.ready()`
+#### `feedStore.ready() -> Promise`
 
 Wait for feedStore to be ready.
 
@@ -174,6 +174,18 @@ Filter the opened feeds using a callback function.
 
 - `descriptor: FeedDescriptor`
 
+#### `feedStore.createReadStream([options]) -> ReadableStream`
+
+Creates a ReadableStream from the feeds stored in FeedStore.
+
+- `options: Options for the hypercore.createReadStream`
+
+#### `feedStore.createReadStreamByFilter(descriptor => Boolean, [options]) -> ReadableStream`
+
+Creates a ReadableStream from the feeds stored in FeedStore.
+
+- `options: Options for the hypercore.createReadStream`
+
 ### Events
 
 #### `feedStore.on('ready', () => {})`

package.json

@@ -25,20 +25,25 @@
     "coverage": "npm test -- --coverage",
     "coveralls": "npm run coverage && cat ./coverage/lcov.info | coveralls",
     "lint": "semistandard 'src/**/*.js'",
-    "prepublishOnly": "npm run test && npm run build",
-    "test": "npm run build:protobuf && jest --verbose --passWithNoTests src",
+    "prepublishOnly": "npm run build && npm run test",
+    "test": "jest --verbose --passWithNoTests src",
     "posttest": "npm run lint"
   },
   "browserslist": [
     "> 5%"
   ],
+  "jest": {
+    "testEnvironment": "node"
+  },
   "dependencies": {
     "@dxos/codec-protobuf": "^1.0.0",
     "buffer-json-encoding": "^1.0.2",
     "debug": "^4.1.1",
+    "end-of-stream": "^1.4.4",
     "hypercore": "^7.7.1",
     "hypercore-crypto": "^1.0.0",
     "hypertrie": "^3.8.0",
+    "multi-read-stream": "^2.0.0",
     "mutexify": "^1.2.0",
     "p-timeout": "^3.2.0",
     "pify": "^4.0.1",
@@ -57,10 +62,12 @@
     "babel-plugin-inline-import": "^3.0.0",
     "coveralls": "^3.0.7",
     "del-cli": "^3.0.0",
+    "end-of-stream-promise": "^1.0.0",
     "jest": "^24.5.0",
     "random-access-memory": "^3.1.1",
     "semistandard": "^14.2.0",
-    "tempy": "^0.3.0"
+    "tempy": "^0.3.0",
+    "wait-for-expect": "^3.0.1"
   },
   "publishConfig": {
     "access": "public"

src/feed-store.js

@@ -4,6 +4,8 @@
 
 import { EventEmitter } from 'events';
 import assert from 'assert';
+import multi from 'multi-read-stream';
+import eos from 'end-of-stream';
 
 import Codec from '@dxos/codec-protobuf';
 
@@ -174,7 +176,7 @@ class FeedStore extends EventEmitter {
 
   /**
    * Get a descriptor by a path.
-   *
+*
    * @param {string} path
    * @returns {FeedDescriptor}
    */
@@ -195,12 +197,12 @@ class FeedStore extends EventEmitter {
   /**
    * Find a feed using a filter callback.
    *
-   * @param {descriptorCallback} callback
+   * @param {descriptorCallback} callback Filter function for the opened descriptors.
    * @returns {Hypercore}
    */
-  findFeed (cb) {
+  findFeed (callback) {
     const descriptor = this.getOpenedDescriptors()
-      .find(descriptor => cb(descriptor));
+      .find(descriptor => callback(descriptor));
 
     if (descriptor) {
       return descriptor.feed;
@@ -210,27 +212,27 @@ class FeedStore extends EventEmitter {
   /**
    * Filter feeds using a filter callback.
    *
-   * @param {descriptorCallback} callback
+   * @param {descriptorCallback} callback Filter function for the opened descriptors.
    * @returns {Hypercore[]}
    */
-  filterFeeds (cb) {
+  filterFeeds (callback) {
     const descriptors = this.getOpenedDescriptors()
-      .filter(descriptor => cb(descriptor));
+      .filter(descriptor => callback(descriptor));
 
     return descriptors.map(descriptor => descriptor.feed);
   }
 
   /**
    * Load feeds using a filter callback.
    *
-   * @param {descriptorCallback} callback
+   * @param {descriptorCallback} callback Filter function for the opened descriptors.
    * @returns {Promise<Hypercore[]>}
    */
-  async loadFeeds (cb) {
+  async loadFeeds (callback) {
     await this.ready();
 
     const descriptors = this.getDescriptors()
-      .filter(descriptor => cb(descriptor));
+      .filter(descriptor => callback(descriptor));
 
     return Promise.all(descriptors.map(descriptor => this._openFeed(descriptor)));
   }
@@ -344,6 +346,44 @@ class FeedStore extends EventEmitter {
     }
   }
 
+  /**
+   * Creates a ReadableStream from the feeds stored in FeedStore.
+   *
+   * @param {Object} [options] Options for the hypercore.createReadStream.
+   * @returns {ReadableStream}
+   */
+  createReadStream (options) {
+    return this.createReadStreamByFilter(() => true, options);
+  }
+
+  /**
+   * Creates a ReadableStream from multiple feeds using a filter function.
+   *
+   * @param {descriptorCallback} [callback] Filter function to select from which feeds to read.
+   * @param {Object} [options] Options for the hypercore.createReadStream.
+   * @returns {ReadableStream}
+   */
+  createReadStreamByFilter (callback, options) {
+    const streams = [];
+
+    this.filterFeeds(callback).forEach(feed => {
+      streams.push(feed.createReadStream(options));
+    });
+
+    const multiReader = multi.obj(streams);
+
+    const onFeed = (feed, descriptor) => {
+      if (callback(descriptor)) {
+        multiReader.add(feed.createReadStream(options));
+      }
+    };
+
+    this.on('feed', onFeed);
+    eos(multiReader, () => this.removeListener('feed', onFeed));
+
+    return multiReader;
+  }
+
   /**
    * Factory to create a new FeedDescriptor.
    *