feat: streams in browser and Bzz API revamp

__tests__/api-bzz-node.ts

@@ -4,6 +4,8 @@
 
 import * as os from 'os'
 import * as path from 'path'
+import { Readable } from 'stream'
+import * as crypto from 'crypto'
 import * as fs from 'fs-extra'
 import { Subject } from 'rxjs'
 import * as tar from 'tar-stream'
@@ -68,6 +70,30 @@ describe('api-bzz-node', () => {
     expect(await response.text()).toBe(uploadContent)
   })
 
+  it('uploading and downloading single file using bzz and streams', async () => {
+    const value = crypto.randomBytes(60).toString('hex')
+    const s = new Readable()
+    s.push(value)
+    s.push(null)
+
+    const manifestHash = await bzz.upload(s, {
+      contentType: 'text/plain',
+    })
+
+    const data: Array<Uint8Array> = []
+    const responseStream = await bzz.downloadStream(manifestHash)
+    responseStream.on('data', (d: Uint8Array) => {
+      data.push(d)
+    })
+
+    return new Promise(resolve => {
+      responseStream.on('end', () => {
+        expect(Buffer.concat(data).toString()).toBe(value)
+        resolve()
+      })
+    })
+  })
+
   it('uploading and downloading single file using bzz with content path', async () => {
     const manifestHash = await bzz.upload(uploadContent, {
       contentType: 'text/plain',
@@ -84,6 +110,30 @@ describe('api-bzz-node', () => {
     expect(await response.text()).toBe(uploadContent)
   })
 
+  it('uploading and downloading single file using bzz-raw and streams', async () => {
+    const value = crypto.randomBytes(60).toString('hex')
+    const s = new Readable()
+    s.push(value)
+    s.push(null)
+
+    const manifestHash = await bzz.upload(s, { size: value.length })
+
+    const data: Array<Uint8Array> = []
+    const responseStream = await bzz.downloadStream(manifestHash, {
+      mode: 'raw',
+    })
+    responseStream.on('data', (d: Uint8Array) => {
+      data.push(d)
+    })
+
+    return new Promise(resolve => {
+      responseStream.on('end', () => {
+        expect(Buffer.concat(data).toString()).toBe(value)
+        resolve()
+      })
+    })
+  })
+
   it('downloading the manifest', async () => {
     const manifestHash = await bzz.upload(uploadContent, {
       contentType: 'text/plain',

__tests__/browser.ts

@@ -16,16 +16,34 @@ describe('browser', () => {
     page.on('console', msg => {
       for (let i = 0; i < msg.args().length; ++i) {
         /* eslint-disable-next-line no-console */
-        console.log(`${i}: ${msg.args()[i]}`)
+        console.log(msg.args()[i])
       }
     })
 
+    page.on('pageerror', function(err) {
+      /* eslint-disable-next-line no-console */
+      console.log('Page error: ' + err.toString())
+    })
+
+    page.on('error', function(err) {
+      /* eslint-disable-next-line no-console */
+      console.log('Error: ' + err.toString())
+    })
+
     await page.addScriptTag({
       path: resolve(
         __dirname,
         '../packages/swarm-browser/dist/erebos.swarm.development.js',
       ),
     })
+
+    await page.addScriptTag({
+      path: resolve(
+        __dirname,
+        '../packages/swarm-browser/dist/readable-stream.js',
+      ),
+    })
+
     await page.addScriptTag({
       path: resolve(
         __dirname,
@@ -134,6 +152,39 @@ describe('browser', () => {
       expect(evalResponse).toBe(uploadContent)
     })
 
+    it('uploads/downloads the file with bzz using streams', async () => {
+      const manifestHash = await evalClient(async (client, uploadContent) => {
+        const s = new NodeStream.Readable()
+        s.push(uploadContent)
+        s.push(null)
+
+        return await client.bzz.upload(s, {
+          contentType: 'text/plain',
+        })
+      }, uploadContent)
+
+      const evalResponse = await evalClient(async (client, manifestHash) => {
+        const response = await client.bzz.downloadStream(manifestHash)
+        return new Promise(resolve => {
+          const data: Array<Uint8Array> = []
+
+          response.on('data', (d: Uint8Array) => {
+            data.push(d)
+          })
+
+          response.on('end', () => {
+            resolve(data)
+          })
+        })
+      }, manifestHash)
+
+      // Reconstruct Buffer
+      const decodedResponse = evalResponse.map(b =>
+        Buffer.from(Object.values(b)),
+      )
+      expect(Buffer.concat(decodedResponse).toString()).toBe(uploadContent)
+    })
+
     it('lists common prefixes for nested directories', async () => {
       const expectedCommonPrefixes = ['dir1/', 'dir2/']
       const dirs = {
@@ -227,6 +278,31 @@ describe('browser', () => {
       expect(directoryList).toEqual({ ...dir, '/': dir[defaultPath] })
     })
 
+    it('downloadDirectoryData() streams the same data provided to uploadDirectory()', async () => {
+      const dir = {
+        [`foo-${uploadContent}.txt`]: {
+          data: `this is foo-${uploadContent}.txt`,
+        },
+        [`bar-${uploadContent}.txt`]: {
+          data: `this is bar-${uploadContent}.txt`,
+        },
+      }
+
+      const downloadedDir = await evalClient(async (client, dir) => {
+        const dirHash = await client.bzz.uploadDirectory(dir)
+        const response = await client.bzz.downloadDirectoryData(dirHash)
+        return Object.keys(response).reduce(
+          (prev, current) => ({
+            ...prev,
+            [current]: { data: response[current].data.toString('utf8') },
+          }),
+          {},
+        )
+      }, dir)
+
+      expect(downloadedDir).toEqual(dir)
+    })
+
     it('supports feeds posting and getting', async () => {
       jest.setTimeout(20000)
       const data = { test: uploadContent }

docs/api-bzz.md

@@ -407,13 +407,33 @@ The `download()` method returns a [`Response` instance](https://developer.mozill
 
 **Returns** `Promise<Response>`
 
+### .downloadStream()
+
+The `downloadStream()` method returns a NodeJs's compatible [`Readable` instance](https://nodejs.org/api/stream.html#stream_class_stream_readable) if the request succeeds, or throws a `HTTPError`.
+
+**Arguments**
+
+1.  `hashOrDomain: string`: ENS name or Swarm hash
+1.  [`options?: DownloadOptions = {}`](#downloadoptions) optional object providing the `path`, `mode` and `contentType`.
+
+**Returns** `Promise<Readable>`
+
+### .downloadDirectoryData()
+
+**Arguments**
+
+1.  `hashOrDomain: string`: ENS name or Swarm hash
+1.  [`options?: DownloadOptions = {}`](#downloadoptions)
+
+**Returns** `Promise<DirectoryData>`
+
 ### .uploadFile()
 
 Uploads a single file and returns the hash. If the `contentType` option is provided, it will return the manifest hash, otherwise the file will be uploaded as raw data and will return the hash of the data itself.
 
 **Arguments**
 
-1.  `data: string | Buffer`
+1.  `data: string | Buffer | Readable`
 1.  [`options: UploadOptions = {}`](#uploadoptions)
 
 **Returns** `Promise<string>`
@@ -667,15 +687,6 @@ Returns a [RxJS `Observable`](https://rxjs.dev/api/index/class/Observable) emitt
 
 **Returns** `Observable<FileEntry>`
 
-### .downloadDirectoryData()
-
-**Arguments**
-
-1.  `hashOrDomain: string`: ENS name or Swarm hash
-1.  [`options?: DownloadOptions = {}`](#downloadoptions)
-
-**Returns** `Promise<DirectoryData>`
-
 ### .downloadTarTo()
 
 **Arguments**
@@ -718,15 +729,6 @@ Call `downloadFileTo()` or `downloadDirectoryTo()` depending on the provided `pa
 
 **Returns** `Promise<void>`
 
-### .uploadFileStream()
-
-**Arguments**
-
-1.  `stream: Readable`: Node.js [`Readable stream`](https://nodejs.org/dist/latest-v10.x/docs/api/stream.html#stream_class_stream_readable) instance
-1.  [`options?: UploadOptions = {}`](#uploadoptions)
-
-**Returns** `Promise<string>`
-
 ### .uploadTar()
 
 **Arguments**

packages/api-bzz-base/package.json

@@ -27,6 +27,11 @@
     "@babel/runtime": "^7.6.2",
     "@erebos/hex": "^0.10.0",
     "@erebos/keccak256": "^0.10.0",
-    "rxjs": "^6.5.3"
+    "readable-stream": "^3.1.1",
+    "rxjs": "^6.5.3",
+    "tar-stream": "^2.1.0"
+  },
+  "devDependencies": {
+    "@types/readable-stream": "^2.3.5"
   }
 }