streamr-dev / broker

A full-featured broker node implementation for the Streamr Network. Supports data persistence via Apache Cassandra and foreign protocol support for MQTT, WebSockets, and HTTP.

NOTICE! This repository has been moved under network monorepo

streamr-broker

Main executable for running a broker node in Streamr Network.

The broker node extends the minimal network node provided by the streamr-network library with

• client-facing support for foreign protocols (e.g. HTTP, MQTT) via adapters
• support for long-term persistence of data using Apache Cassandra.

Table of Contents

• Install
• Run
• Develop
• Release
• Misc

Install

Prerequisites are Node.js 14.x and npm version >=6.14.

To install streamr-broker:

npm install streamr-broker --global

Run

It is convenient to run a broker node as part of the full Streamr stack. Check out the streamr-docker-dev tool that can be used to run the full Streamr stack.

If instead you want to run a broker node by itself without Docker, follow the steps below.

First install the package

npm install streamr-broker --global

Then run the command broker with the desired configuration file

broker <configFile>

See folder "configs" for example configurations. To run a simple local broker

broker configs/development-1.env.json

Then run the command tracker with default values

tracker

Deleting expired data from Storage node

To delete expired data from storage node run

broker <configFile> --deleteExpired

or

node app.js <configFile> --deleteExpired

Develop

Install dependencies:

npm ci

Run the tests:

npm run test

We use eslint for code formatting:

npm run eslint

Code coverage:

npm run coverage

Debug

To get all debug messages:

LOG_LEVEL=debug

... or adjust debugging to desired level:

LOG_LEVEL=[debug|info|warn|error]

To disable all logs

NOLOG=true

Regenerate self-signed certificate fixture

To regenerate self-signed certificate in ./test/fixtures run:

openssl req -x509 -newkey rsa:4096 -keyout key.pem -out cert.pem -days 36500 -nodes -subj "/CN=localhost"

Release

Publishing to NPM is automated via GitHub Actions. Follow the steps below to publish.

1. git checkout master && git pull
2. Update version with either npm version patch, npm version minor, or npm version major. Use semantic versioning https://semver.org/. Files package.json and package-lock.json will be automatically updated, and an appropriate git commit and tag created.
3. git push --follow-tags
4. Wait for GitHub Actions to run tests
5. If tests passed, GitHub Actions will publish the new version to NPM

Misc

API Specification

For production version of the Streamr API refer to API Explorer.

Special considerations for using MQTT adapter

• For authentication put private key in the password connection field
• MQTT clients can send plain text, but their payload will be transformed to a JSON object accordingly: {"mqttPayload":"ORIGINAL_PLAINTEXT_PAYLOAD}

Error handling

• If private key is not correct, client will receive "Connection refused, bad user name or password" (returnCode: 4)
• If stream is not found, client will receive "Connection refused, not authorized" (returnCode: 5)

broker's Issues

Write a test for Mqtt error handling

Write a (integration) test for the scenario fixed by #24 .

Can't install Broker on Ubuntu 16 / Node v16.1.0

The command npm install streamr-broker --global breaks with the error:

path /usr/lib/node_modules/streamr-broker/node_modules/node-datachannel
npm ERR! command failed

Error: Not supported

Sentry Issue: NETWORK-H

Error: Not supported
  File "/srv/broker/node_modules/mqtt-packet/parser.js", line 121, in Parser._parsePayload
    this._emitError(new Error('Not supported'))
  File "/srv/broker/node_modules/mqtt-packet/parser.js", line 39, in Parser.parse
    this[this._states[this._stateCounter]]() &&
  File "/srv/broker/node_modules/mqtt-connection/lib/parseStream.js", line 14, in DestroyableTransform.process [as _transform]
    that.parser.parse(chunk)
  File "/srv/broker/node_modules/readable-stream/lib/_stream_transform.js", line 184, in DestroyableTransform.Transform._read
    this._transform(ts.writechunk, ts.writeencoding, ts.afterTransform);
  File "/srv/broker/node_modules/readable-stream/lib/_stream_transform.js", line 172, in DestroyableTransform.Transform._write
    if (ts.needTransform || rs.needReadable || rs.length < rs.highWaterMark) this._read(rs.highWaterMark);
...
(5 additional frame(s) were not displayed)

Add more integration tests for resend cancellation

Maybe one more tests when you have 2 or 3 resend requests within one broker (we had before problems where we where testing only 1 instance instead of N)?
And one more test between two brokers and check that when client disconnects, all brokers actually cancel resend requests.

Rename StreamFetcher to EngineAndEditorClient or something similar

StreamFetcher now contains more logic than just fetching stream information from E&E. It also contains methods for fetching other resources.

Error: Cannot parse protocolId

Sentry Issue: NETWORK-P

Error: Cannot parse protocolId
  File "/home/ubuntu/.nvm/versions/node/v12.7.0/lib/node_modules/@streamr/broker/node_modules/mqtt-packet/parser.js", line 143, in Parser._parseConnect
    if (protocolId === null) return this._emitError(new Error('Cannot parse protocolId'))
  File "/home/ubuntu/.nvm/versions/node/v12.7.0/lib/node_modules/@streamr/broker/node_modules/mqtt-packet/parser.js", line 84, in Parser._parsePayload
    this._parseConnect()
  File "/home/ubuntu/.nvm/versions/node/v12.7.0/lib/node_modules/@streamr/broker/node_modules/mqtt-packet/parser.js", line 39, in Parser.parse
    this[this._states[this._stateCounter]]() &&
  File "/home/ubuntu/.nvm/versions/node/v12.7.0/lib/node_modules/@streamr/broker/node_modules/mqtt-connection/lib/parseStream.js", line 14, in DestroyableTransform.process [as _transform]
    that.parser.parse(chunk)
  File "/home/ubuntu/.nvm/versions/node/v12.7.0/lib/node_modules/@streamr/broker/node_modules/readable-stream/lib/_stream_transform.js", line 184, in DestroyableTransform.Transform._read
    this._transform(ts.writechunk, ts.writeencoding, ts.afterTransform);
...
(5 additional frame(s) were not displayed)

Solve --forceExit in integration tests

Integration tests hang after running all tests. Some unclosed handles.

Error: Cannot parse topic

Sentry Issue: NETWORK-4

Error: Cannot parse topic
  File "/srv/broker/node_modules/mqtt-packet/parser.js", line 259, in Parser._parsePublish
    if (packet.topic === null) return this._emitError(new Error('Cannot parse topic'))
  File "/srv/broker/node_modules/mqtt-packet/parser.js", line 90, in Parser._parsePayload
    this._parsePublish()
  File "/srv/broker/node_modules/mqtt-packet/parser.js", line 39, in Parser.parse
    this[this._states[this._stateCounter]]() &&
  File "/srv/broker/node_modules/mqtt-connection/lib/parseStream.js", line 14, in DestroyableTransform.process [as _transform]
    that.parser.parse(chunk)
  File "/srv/broker/node_modules/readable-stream/lib/_stream_transform.js", line 184, in DestroyableTransform.Transform._read
    this._transform(ts.writechunk, ts.writeencoding, ts.afterTransform);
...
(5 additional frame(s) were not displayed)