Redux Real-time CQRS

Redux Real-time CQRS is a javascript library that implements CQRS pattern and real-time updates for applications developed with React (React-native) and redux

Motivation

The react-flux combination sounds at first a relationship that can simplify the development in an unprecedented way, but when application grows up, maintainance may be really hard because the lack of structure and responsibility segregation

Among other things, the weaknesses of a large application developed with react and flux can be:

• Most developers who are coding with flux are not clear on how to organize the file structure
• There is no Single Responsability Principle on flux, ie, Where is the best place to put the logic? The action, the store, the component, Which one of those are the best for putting logic?
• There is no clear division of responsibilities. Maintainability can be difficult and increase technical debt with each update, because the developers are not clear about where the correct place is for logic or mutation
• On the other hand, most of the available literature does not give us a guideline to interact with the server, either request/response or publisher/subscriber. How to do it in a transparent and maintainable way?
• And finally, there is a growing need to convert our traditional request/response applications into real-time applications with the lowest possible cost, preferably without change language, database or technologies that are used, which it is almost impossible!

CQRS Middleware

Architecture

As illustrated in the graph there are new important concepts that are present on cqrsMiddleware, They are explained below:

Event

An Event is a specialization of an Action whose unique responsability is to update the state, these Events must always be named in past tense. An Event is something that imminently must modify the state. It is assumed, for an Action becomes Event, the Event must have gone through a validation process if necessary. Ejm:

    class ProjectAddedEvt extends IdentifiedAction {

        name:string;
        status:string;
        timestamp:number;
        tempId:string;

        constructor(id:string, name:string, status:string, timestamp:number, tempId:string) {
            super(id);
            this.name = name;
            this.status = status;
            this.timestamp = timestamp;
            this.tempId = tempId;
        }
    }

In this example, the ProjectAddedEvt inherits from IdentifiedAction which is nothing more than an Action with an id

Command

A Command, on the other hand, is the specialization of an Action which will trigger a Handler (a piece of logic) associated with Command. Ejm:

    class AddProjectCmd extends Action {
        name:string;

        constructor(name) {
            super();
            this.name = name;
        }
    }

For creation of a Project it's not necessary the id, therefore AddProjectCmd inherits simply from Action

Handler

A Handler is associated with one or more Command and vice versa, and each Handler is executed when the Command (associated with the Handler) is dispatched Ejm:

    @HandlerOf([SomeCommand])
    class SomeCommandHandler {
        static run(dispatch, action, state) {

            console.log("this is a test")

        }
    }

@HandlerOf

@HandlerOf Decorator is needed to associate one or more Command to any Handler, cqrsMiddleware will search for the Handlers associated with the dispatched Command

run(dispatch, action, state)

Static function which is executed by the cqrsMiddleware once a Command is dispatched

dispatch

If necessary, you can change the state after executing a Handler using the dispatch function with an Event, the Event will go directly to the reducers because this Action (Event) is not associated with any Handler . There is also the possibility that a Handler dispatches a Command, in which case another Handler will be executed serially. Ejm:

    @HandlerOf([FindProjectByIdCmd])
    class FindProjectByIdCmdHandler {
        static run(dispatch, action, state) {

            fetch(`http://localhost:9000/projects/${action.id}`)
                .then(function (response) {
                    response.json().then((json)=> {
                        let data = json.data;
                        let project = new ProjectAddedEvt(data.id, data.name, data.status, data.timestamp, data.id);
                        dispatch(project.toPlainJSON())
                    });
                });

        }
    }

In this case, when you dispatch a Command with type FindProjectByIdCmd , the middleware will call run method of FindProjectByIdCmdHandler. Once you have communicated to the server, the Handler will dispatch an Event with type ProjectAddedEvt. Finally the state changes because the subscribed reducer associated with ProjectAddedEvt (Use toPlainJSON () to convert an Action into a plain JSON ) ProjectAddedEvt reducer:

    function projects(state = [], action = {}) {
        switch (action.type) {
            case ProjectAddedEvt.name:
                let newState = [
                    ...[...state].filter(item=>item.id != action.id && (!item.tempId || item.tempId != action.tempId)),
                    action
                ];
                newState.sort((a, b)=>b.timestamp - a.timestamp);
                return newState;
            default:
                return state;
        }
    }

action

It's the Command instance associated with the Handler. Take note that this action is a plain JSON

state

Read-only object that represents the current application state

return

In case Handler has return statement, Ejm:

    @HandlerOf([ToggleTaskCmd])
    class ToggleTaskCmdHandler {
        static run(dispatch, action, state) {

            let task = state.tasks.find(task=>task.id === action.id);
            fetch(`http://localhost:9000/tasks/${action.id}/toggle`, {
                method: 'put',
                headers: new Headers({
                    'Content-Type': 'application/json'
                })
            }).then(function (response) {
                response.json().then((json)=> {
                    //id:string, name:string, status:string, timestamp:number, completed:boolean, tempId:string
                    dispatch(new TaskAddedEvt(task.id, task.name, Constants.SERVER_READY, task.timestamp, !task.completed, task.tempId).toPlainJSON())
                });
            });
            return new TaskAddedEvt(task.id, task.name, Constants.SERVER_PENDING, task.timestamp, !task.completed, task.tempId).toPlainJSON()
        }
    }

The returned value will be sent through the dispatch, ie, It will be delivered immediately later the return statement

cqrsMiddleware([Handler1, ...HandlerN])

This is the function that returns the middleware used by applyMiddleware. This function receives a list of Handlers that you want to subscribe for the application, the Handlers that are not placed in this array will not be executed Ejm:

    let commandHandlers = [
        FindProjectsCmdHandler,
        SelectProjectCmdHandler,
        AddProjectCmdHandler,
        AddTaskCmdHandler,
        ToggleTaskCmdHandler,
        FindProjectByIdCmdHandler,
        FindTaskByIdCmdHandler,
        DeleteTaskByIdCmdHandler,
        GetTasksProjectCmdHandler,
        NotifyProjectDeletedCmdHandler,
        DeleteProjectByIdCmdHandler
    ];


    const middleware = [cqrsMiddleware(commandHandlers), realTimeUpdatingMiddleware(config)];
    const store = compose(
        applyMiddleware(...middleware),
        devTools()
    )(createStore)(reducers);

Real-time Middleware

If you have integrated cqrsMiddleware to your redux application, the implementation of real-time updates is almost transparent. The only thing necessary is having a real-time bus for notifying our application about changes on entities or streams, we use Firebase as our real-time bus

Rest Resource

Concepts

Entity

An Entity is a rest resource which has id

Ejm: GET http://localhost:9000/projects/1

    {
      "data": {
        "name": "First Project",
        "id": 1,
        "timestamp": 1,
        "status": "SERVER_READY"
      },
      "timestamp": 1464122123739031
    }

Stream

It's a list of Entities with undeterminated size, which belogs to another Entity. Usually these lists are paginated

Ejm GET http://localhost:9000/projects/1/tasks

    {
      "data": [
        {
          "name": "First todo of first project",
          "completed": false,
          "projectId": 1,
          "id": 1,
          "timestamp": 1,
          "status": "SERVER_READY"
        },
        {
          "name": "Second todo of first project",
          "completed": false,
          "projectId": 1,
          "id": 2,
          "timestamp": 2,
          "status": "SERVER_READY"
        }
      ],
      "timestamp": 1464122245701952
    }

Real-time Bus

It's the key part of converting a traditional API Rest into a real-time API, the bus tells our application when to update an Entity or Stream. For example: the app will execute a new GET request when an specific resource has been updated and notified by Firebase. The backend side have to update the Firebase database with the same structure of the rest resource Ejm:

With this structure being managed from the backend, realTimeMiddleware knows exactly when to update, add, or delete data on application state, this thanks to the timestamp which change (on Firebase) on every update, delete or when an item is added or deleted on a stream on server side

Usage

For converting your application into a real-time application (listening Firebase for updates) just decorate your Event with @RealTime. Ejm:

@RealTime("projects", FindProjectByIdCmd, NotifyProjectDeletedCmd, [["tasks", GetTasksProjectCmd]])
    class ProjectAddedEvt extends IdentifiedAction {
        name:string;
        status:string;
        timestamp:number;
        tempId:string;

        constructor(id:string, name:string, status:string, timestamp:number, tempId:string) {
            super(id);
            this.name = name;
            this.status = status;
            this.timestamp = timestamp;
            this.tempId = tempId;
        }
    }

Every time an Action is annotated with @RealTime, and the instance of this Action is dispatched, the realTimeMiddlware is notified to listen changes on Firebase. Details of decorator are discribed bellow

@RealTime(path:string, onUpdate:Action, onDelete:Action, onUpdateStream:Array<[string, Action]>)

The decorator takes 4 parameters for reacting to changes on Firebase. Every Action annotated with @RealTime implies there is a new item on application state, cqrsMiddleware will listen each time firebase change on the specific item URI (in our example "/projects/1/timestamp") or associated streams ("/projects/1/tasks/timestamp")

path:string

The first parameter is the path which cqrsMiddleware will listen (in our example "/projects"), this path is concatenated with the id of the dispatched Action (so the Action must extend from IdentifiedAction) to create a callback on the concatenated path (Ejm: "/projects/:id")

onUpdate:Action

Whenever there is a change on the timestamp field of "path/id" on Firebase (Ejm "projects/1/timestamp"), the onUpdate parameter (inherited from IdentifiedAction because has id) will be sent to the dispatcher (dispatch(onUpdate)). For example:

    @RealTime("projects", FindProjectByIdCmd, NotifyProjectDeletedCmd, [["tasks", GetTasksProjectCmd]])
    class ProjectAddedEvt extends IdentifiedAction {

        name:string;
        status:string;
        timestamp:number;
        tempId:string;

        constructor(id:string, name:string, status:string, timestamp:number, tempId:string) {
            super(id);
            this.name = name;
            this.status = status;
            this.timestamp = timestamp;
            this.tempId = tempId;
        }
    }

In the example, whenever there is a change on /projects/1/timestamp on Firebase, the FindProjectByIdCmd command will be dispatched. For this event (FindProjectByIdCmd) there is a Handler associated:

    @HandlerOf([FindProjectByIdCmd])
    class FindProjectByIdCmdHandler {
        static run(dispatch, action, state) {

            fetch(`http://localhost:9000/projects/${action.id}`)
                .then(function (response) {
                    response.json().then((json)=> {
                        let data = json.data;
                        let project = new ProjectAddedEvt(data.id, data.name, data.status, data.timestamp, data.id);
                        dispatch(project.toPlainJSON())
                    });
                });

        }
    }

Each time the field /projects/1/timestamp is modified on Firebase, the application will execute a GET Request to http://localhost:9000/projects/${action.id} because of FindProjectByIdCmdHandler

onDelete:Action

Similarly, whenever the "/projects/1/timestamp" field is deleted on Firebase, the command NotifyProjectDeletedCmd will be dispatched, the command should have a Handler associated:

    @HandlerOf([NotifyProjectDeletedCmd])
    class NotifyProjectDeletedCmdHandler {
        static run(dispatch, action, state) {
            if (state.selectedProjectId === action.id) {

                Actions.projectList()

            }
            dispatch(new FindProjectsCmd().toPlainJSON())
        }
    }

The example is using the react-native-redux-router library, so every time a project item is deleted on Firebase, the application will change the route to projectList and will dispatch the command FindProjectsCmd

onUpdateStream:Array<[string, Action]>

Finally, if the entity has associated streams (In our example tasks, because every Project has a list of Task), the onUpdateStream parameter allows you to define which commands will be dispatched each time the server add or delete an item to the stream (tasks), repesented by the timestamp on Firebase.

In our example: every time the value of field /projects/1/tasks/timestamp on Firebase (that is, server has added or deleted an item to "tasks" stream) changes, the command GetTasksProjectCmd will be dispatched to the following Handler:

    @HandlerOf([GetTasksProjectCmd])
    class GetTasksProjectCmdHandler {
        static run(dispatch, action, state) {

            if (state.selectedProjectId == action.id) {
                state.tasks.forEach((task)=>dispatch(new TaskDeletedEvent(task.id).toPlainJSON()));
                fetch(`http://localhost:9000/projects/${action.id}/tasks`)
                    .then(function (response) {
                        response.json().then((json)=> {
                            //id:string, name:string, status: string, timestamp:number,tempId: string
                            Object.keys(json.data)
                                .map(key=>json.data[key]).map(item=>new TaskAddedEvt(item.id, item.name, item.status, item.timestamp, item.completed, item.id).toPlainJSON()).forEach(evt=>dispatch(evt))
                        });
                    });
            }
        }
    }

This specific Handler executes a new request to http://localhost:9000/projects/${action.id}/tasks