Connect to Kafka via Lenses

Redux middleware that facilitates connection to a kafka backend running Landoop's Lenses.(Tested with v3.7)

Table of contents

• Install
• Usage
• Options
• To Do
• Notes
• License

Install

npm i --save redux-lenses-streaming rxjs

The only peer dependency that the library has is rxjs.

Usage

First setup the redux store with the kafka middleware:

import { createLensesMiddleware } from 'redux-lenses-streaming';

function configureStore() {
  const lensesMiddleware = createLensesMiddleware();
  //Any other middleware you might use
  const logger = createLogger();
  const middleware = [logger, lensesMiddleware];

  const store = createStore(
    rootReducer,
    applyMiddleware(...middleware),
  );

  return store;
}

Or you can customise the middleware with custom options in which case it will attempt to connect:

import { createLensesMiddleware } from 'redux-lenses-streaming';

function configureStore() {
  const lensesOptions = {
    host: 'cloudera02.landoop.com:24006/api/kafka/ws',
    clientId: 'MyClientsName',
    // See Options section for full list
  };

  const lensesMiddleware = createLensesMiddleware(lensesOptions);
  const middleware = [..., lensesMiddleware];

  const store = createStore(
    rootReducer,
    applyMiddleware(...middleware),
  );

  return store;
}

After, add the reducer:

import { combineReducers } from 'redux';
import { lensesReducer } from 'redux-lenses-streaming';
import sessionReducer from './sessionReducer';

const rootReducer = combineReducers({
  lenses: lensesReducer,
  //Add other application reducers where you listen
  // to the exposed actions (see list below)
  session: sessionReducer,
});

export default rootReducer;

Now you are ready to dispatch Actions using the provided action creators, that the middleware will intercept:

import { Actions } from 'redux-lenses-streaming';

dispatch(Actions.connect(options));
dispatch(Actions.login(options));
dispatch(Actions.publish(payload));
dispatch(Actions.subscribe(payload));
dispatch(Actions.unsubscribe(payload));
dispatch(Actions.disconnect());
...

You can also listen to various Action Types, dispatched by the middleware:

import { Type } from 'redux-lenses-streaming';

export const Type = {
  KAFKA_MESSAGE
  KAFKA_HEARTBEAT
  CONNECT
  CONNECT_SUCCESS
  CONNECT_FAILURE
  DISCONNECT
  DISCONNECT_SUCCESS
  DISCONNECT_FAILURE
  LOGIN
  LOGIN_SUCCESS
  LOGIN_FAILURE
  PUBLISH
  PUBLISH_SUCCESS
  PUBLISH_FAILURE
  SUBSCRIBE
  SUBSCRIBE_SUCCESS
  SUBSCRIBE_FAILURE
  UNSUBSCRIBE
  UNSUBSCRIBE_SUCCESS
  UNSUBSCRIBE_FAILURE
};
...

Options

Passed when creating middleware or when dispatching connect action.

const defaultOptions = {
  host: '',
  clientId: '',
  user: '',
  password: '',
  secure: false,
  timeout: 5000,
  autoCommitDelay: -1,
};

// For login action creator:
const options = {
  user: '',
  password: '',
};

host (String)

Web socket address, including port. If wss:// is not set, it will be added by the library. Example of address: test.landoop.com:21112/api/kafka/ws wss://test.landoop.com:21112/api/kafka/ws

Default:

clientId (String)

Client Id. If previous session found, it will send back messages on topic subscription.

Default:

user (String)

User for authentication.

Default:

password (String)

Password for authentication.

Default:

timeout (Integer)

Timeout (ms) before publish / subscribe fail.

Default: 5000

autoCommitDelay (Integer)

Delay (ms) for autocommiting last message. If -1, then you will need to manually send commit message.

Default: -1

secure (Boolean)

Force connection to wss.

Default: false

To Do

• [ ] Add Angular implementation
• [ ] Support message batching
• [ ] Handle unexpected disconnects and reconnection attempt
• [ ] Add delay options
• [ ] Tests

Notes

• Tested with Redux v3.7

License

All copyrights reserved - Landoop LTD 2016-2018