@pestras/microservice v1.0.1

Pestras Microservice

Pestras Microservice as PMS is built on nodejs framework using typescript, supporting http rest service, nats server, socket io and can run multible instances based on nodejs cluster with messageing made easy between workers.

Template

$ git clone https://github.com/pestras/pestras-microservice-template

Creating Service

In order to create our service we need to use SERVICE decorator which holds the main configuration of our service class.

import { SERVICE } from '@pestras/microservice';

@SERVICE({ version: 1 })
class Test {}

Service Configurations

LOGLEVEL Enum

PMS provides only four levels of logs grouped in an enum type LOGLEVEL

• LOGLEVEL.ERROR
• LOGLEVEL.WARN
• LOGLEVEL.INFO
• LOGLEVEL.DEBUG

SocketIOOptions

Cors

PM default cors options are:

'access-control-allow-methods': "GET,HEAD,OPTIONS,PUT,PATCH,POST,DELETE",
'access-control-allow-origin': "*",
'access-control-allow-headers': "*",
'Access-Control-Allow-Credentials': 'false',
'success-code': '204'

To change that, overwrite new values into cors options

@SERVICE({
  version: 1,
  cors: {
    'access-control-allow-methods': "GET,PUT,POST,DELETE",
    'access-control-allow-headers': "content-type"
  }
})
class Test {}

Micro

Before delving into service routes, subjects.. etc, let's find out how to run our service..

After defining our service class we use the Micro object to run our service through the start method.

import { SERVICE, Micro } from '@pestras/microservice';

@SERVICE({
  // service config
})
export class TEST {}

Micro.start(Test);

Micro.start method accepts an additionl optional array of sub services, which will be explained later on;

Micro object has another properties and methods that indeed we are going to use as well later in the service.

ROUTE DECORATOR

Used to define a route for a rest service.

ROUTE accepts an optional config object to configure our route.

import { SERVICE, ROUTE } from '@pestras/microservice';

@SERVICE({
  version: 1
})
class Articles {

  @ROUTE({
    // /articles/v1/{id}
    path: '/{id}'
  })
  getArticle(req: Request, res: Response) {
    let id = req.params.id;

    // get article code

    res.json(article);
  }
}

Request

PMS http request holds the original Node IncomingMessage with a few extra properties.

Request Path Patterns

PM path patterns are very useful that helps match specific cases

1. /articles/{id} - id is a param name that match any value: /articles/4384545 or /articles/45geeFEe8 but not /articles or /articles/dsfge03tG9/1

2. /articles/{id}? - same the previous one but id params is optional, so /articles is acceptable.

3. /articles/{cat}/{start}?/{limit}? - cat params is required, however start and limit are optionals, /articles/scifi, /articles/scifi/0, /articles/scifi/0/10 all matched

4. /articles/{id:^0-9{10}$} - id param is constrained with a regex that allow only number value with 10 digits length only.

5. /articles/* - this route has rest operator which holds the values of the rest blocks of the path separated by '/' as an array, articles/scifi/0/10 does match and request.params'*' equals 'scifi','0','10', however /articles does not match

6. /articles/*? - same as the previous however /articles does match

notes:

• Rest operator accepts preceding parameter but not optional parameters.
• Adding flags to regexp would be /articles/{id:a-z{10}:i}.
• Parameters with Regexp can be optional as will /articles/{id:a-z{10}:i}?
• Parameters can be seperated by fixed value blocks /articles/{aid}/comments/{cid}
• Parameters and rest operator can be seperated by fixed value blocks as well.
• On each request, routes are checked in two steps to enhance performance
  • Perfect match: Looks for the perfect match (case sensetive).
  • By Order: if first step fail, then routes are checked by order they were defined (case insensetive)

@SERVICE()
class AticlesQuery {
  // first to check
  @ROUTE({ path: '/{id}'})
  getById() {}
  
  // second to check
  @ROUTE({ path: '/published' })
  getPublished() {}
  
  /**
   * Later when an incomimg reauest made including pathname as: 'articles-query/v0/Published' with capitalized P
   * first route to match is '/{id}',
   * However when the path name is 'articles-query/v0/published' with lowercased p '/published' as the defined route then
   * the first route to match is '/published' instead of '/{id}'
   */
}

Response

PMS http response holds the original Node Server Response with a couple of methods.

Using response.json() will set 'content-type' response header to 'application/json'. Response will log any 500 family errors automatically.

Response Security headers

PM add additional response headers for more secure environment as follows:

'Cache-Control': 'no-cache,no-store,max-age=0,must-revalidate'
'Pragma': 'no-cache'
'Expires': '-1'
'X-XSS-Protection': '1;mode=block'
'X-Frame-Options': 'DENY'
'Content-Security-Policy': "script-src 'self'"
'X-Content-Type-Options': 'nosniff'

Headers can be overwritten using response.setHeaders method,

HOOK DECORATOR

Hooks are called before the actual request handler, they are helpful for code separation like auth, input validation or whatever logic needed, they could be sync or async returning boolean value.

import { Micro, SERVICE, Request, Response, HOOK, ROUTE, CODES } from '@pestras/microservice';

@SERVICE()
class Test {
  @HOOK()
  async auth(req: Request, res: Response, handlerName: string) {
    const user: User;
  
    // some auth code
    // ...

    if (!user) {
      res.status(CODES.UNAUTHORIZED).json({ msg: 'user not authorized' });
      return false;
    }
  
    req.auth = user;
    return true
  }

  @ROUTE({ hooks: ['auth'] })
  handlerName(req: Request, res: Response) {
    const user = req.auth;
  }
}

Micro.start(Test);

Hooks should handle the response on failure and returning or resolving to false, otherwise PM will check response status and if its not ended, it will consider the situation as a bad request from client that did not pass the hook and responding with BAD_REQUEST code 400.

SUBJECT DECORATOR

Used to subscribe to nats server pulished subjects, and also accepts a subject string as a first argument and an optional config object.

import { SERVICE, SUBJECT, NatsMsg } from '@pestras/microservice';
import { Client, Payload} from 'ts-nats';

@SERVICE({
  version: 1,
  workers: 3,
  nats: { url: 'http://localhost:4222', payload: Payload.JSON }
})
class Email {

  // hooks works with subjects as well
  // arguments are swaped with (nats: Nats.Client, msg: NatsMsg, handlerName: string - name of the subject handler method that called the hook)
  @Hook(5000)
  async auth(nats: Client, msg: NatsMsg, handlerName: string) {
    // if hook failed its purpose should check for msg reply if exists and return false
    if (msg.reply) {
      nats.publish(msg.replay, { error: 'some error' })
      return false
    }

    // otherwise
    return true;
  }

  @SUBJECT('user.insert', {
    hooks: ['auth'],
    options: { queue: 'emailServiceWorker' }
  })
  sendActivationEmail(nats: Client, msg: NatsMsg) {
    let auth = msg.data.auth;
  }

Hooks must return or resolve (async) to true on success or false on failure.

Multible Subjects

Multible subjects can be used on the same handler.

import { SERVICE, SUBJECT, NatsMsg } from '@pestras/microservice';
import { Client, Payload} from 'ts-nats';

interface MsgInput { id: string; email: string }

@SERVICE({
  version: 1,
  nats: { url: 'http://localhost:4222', payload: Payload.JSON }
})
class Email {

  @SUBJECT('emails.new')
  @SUBJECT('emails.reactivate')
  sendActivataionEmail(client: Client, msg: NatsMsg<MsgInput>) {
    // send email
  }
}

SocketIO

PMS provides several decorators to manage our SocketIO server.

CONNECT DECORATOR

This decorator will call the method attached to whenever a new socket has connected, it accepts an optional array of namespaces names, defaults to 'default' which is the main io server instance.

import { SERVICE, CONNECT } from '@pestras/microservice';

@SERVICE()
class Publisher {

  @CONNECT()
  onSocketConnect(io: SocketIO.Servier, socket: SocketIO.Socket) {}

  @CONNECT(['blog'])
  onSocketConnectToBlog(ns: SocketIO.Namespace, socket: SocketIO.Socket) {}
}

RECONNECT DECORATOR

Called whenever a socket reconnect to the namespace or the server.

import { SERVICE, RECONNECT } from '@pestras/microservice';

@SERVICE()
class Publisher {

  @RECONNECT()
  onSocketReconnect(io: SocketIO.Servier, socket: SocketIO.Socket) {}

  @RECONNECT(['blog'])
  onSocketReconnectToBlog(ns: SocketIO.Namespace, socket: SocketIO.Socket) {}
}

HANDSHAKE DECORATOE

Called when a socket establish a coonection for the first time, mostly used for authorization.

It accepts an optional array of namespaces names and defaults to 'defualt'.

import { SERVICE, HANDSHAKE } from '@pestras/microservice';

@SERVICE()
class Publisher {

  @HANDSHAKE()
  handshake(io: SocketIO.Servier, socket: SocketIO.Socket, next: (err?: any) => void) {}

  @HANDSHAKE(['blog'])
  blogHandshake(ns: SocketIO.Namespace, socket: SocketIO.Socket, next: (err?: any) => void) {
    
  }
}

USE DECORATOE

Same as HANDSHAKE decorator.

import { SERVICE, USE } from '@pestras/microservice';

@SERVICE()
class Publisher {

  @USE()
  use(io: SocketIO.Servier, socket: SocketIO.Socket, next: (err?: any) => void) {}

  @USE(['blog'])
  blogUse(ns: SocketIO.Namespace, socket: SocketIO.Socket, next: (err?: any) => void) {}
}

USESOCKET DECORATOE

Used to listen to all socket incoming events.

import { SERVICE, USESOCKET } from '@pestras/microservice';

@SERVICE()
class Publisher {

  @USESOCKET()
  useSocket(io: SocketIO.Servier, packet: SocketIO.Packet, next: (err?: any) => void) {}

  @USESOCKET(['blog'])
  blogUseSocket(ns: SocketIO.Namespace, packet: SocketIO.Packet, next: (err?: any) => void) {}
}

EVENT DECORATOE

Used to listen to a specific event, accepts an event name as a first parameter and an optional array of namespaces for the second defaults to 'default'.

import { SERVICE, EVENT } from '@pestras/microservice';

@SERVICE()
class Publisher {

  @EVENT('userLoggedIn')
  userLoggedIn(io: SocketIO.Servier, socket: SocketIO.Socket, ...args: any[]) {}

  @EVENT('newArticle', ['blog'])
  newArticle(ns: SocketIO.Namespace, socket: SocketIO.Socket, ...args: any[]) {}
}

DISCONNECT DECORATOE

Triggered when a socket disconnect form the namespace or the server.

import { SERVICE, DISCONNECT } from '@pestras/microservice';

@SERVICE()
class Publisher {

  @DISCONNECT()
  socketDisconnected(io: SocketIO.Servier, packet: SocketIO.Packet) {}

  @DISCONNECT(['blog'])
  blogSocketDisconnected(ns: SocketIO.Namespace, socket: SocketIO.Socket) {}
}

Sub Services

PM gives us the ability to modulerize our service into subservices for better code splitting.

SubServices are classes that are defined in seperate modules, then imported to the main service module then passed to Micro.start() method to be implemented.

// comments.service.ts
import { ROUTE, SUBJECT, HOOK, SubServiceEvents } from '@pestras/microservice';

export class Comments implements SubServiceEvents {

  async onInit() {}
  
  @HOOK()
  validate(req, res) { return true }
  
  @ROUTE({ 
    path: '/list' // => /artivles/v0/comments/list
    // auth hook from the main service
    // validate from local service
    hooks: ['auth', 'validate']
  })
  list(req, res) {
    res.json([]);
  }

  @SUBJECT('comment-like')
  like(nats, msg) {
    let sharedValue = Micro.store.someSharedValue;
    // save like
  }
}

// main.ts
import { Micro, SERVICE, HOOk, ROUTE, ServiceEvents } from '@pestras/microservice';
import { Comments} from './comments.service'

@SERVICE()
class Articles {

  onInit() {    
    Micro.store.someSharedValue = "shared value";
  }

  @HOOK()
  async auth(req, res) {
    return true;
  }

  @HOOK()
  validate(req, res) {
    return true;
  }

  @ROUTE({
    path: '/list', // => articels/v0/list
    // both hooks from the main service
    hooks: ['auth', 'validate']
  })
  list(req, res) {
    res.json([]);
  }
}

// pass sub services as an array to the second argument of Micro.start method
Micro.start(Articles, [Comments]);

Serveral notes can be observed from the example:

• Routes paths in sub services are prefixed with the sub service name.
• Local hooks has the priority over main service hooks.
• Subservices have their own events onInit, onReady and onRequest.
• Subjects, SocketIO, process MSG decorators are all supported as well.
• Micro.store is shared among all subServices.

SocketIO namespaces cannot be splitted into several subservices, each subservce must have its own namespace.

Cluster

PMS uses node built in cluster api, and made it easy for us to manage workers communications.

First of all to enable clustering we should set workers number in our service configurations to some value greater than one.

import { SERVICE } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher {}

To listen for a message form another process.

import { SERVICE, MSG } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher {

  @MSG('some message')
  onSomeMessage(data: any) {}
}

To send a message to other processes we need to use Micro.message method, it accepts three parameters.

import { SERVICE, Micro } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher {
  
  // some where in your service
  Micro.message('some message', { key: 'value' });
}

In case of not using a socket io adapter, PMS provide another helper method to manage communications between workers for handling socket io broadcasting using Micro.publish method which accepts SocketIOPublishMessage object.

import { SERVICE, EVENT, Micro } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher {

  @EVENT('ArticleUpdated', ['blog'])
  onArticleUpdate(ns: SocketIO.Namespace, socket: SocketIO.Socket, id: string) {
    socket.to('members').emit('ArticleUpdated', id);
    // publish to other worker socket io
    Micro.publish({
      event: 'ArticleUpdated',
      data: [id],
      namespace: 'blog',
      room: 'members'
    });
  }
}

Also it is made easy to restart all workers or the current one.

import { SERVICE, Micro } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher {

  // some where in our service

  // restarting all workers
  Micro.message('restart all');

  // restarting the current worker
  Micro.message('restart');
}

When restarting all workers, it is going to be one by one process, PMS is going to wait for the new worker to start listening and then will restart the next one.

Attempt helper

When a critical async call needs extra caution to avoid failure, attempt method helps to make multiple calls with wait time between and a timeout.

@SERVICE()
class Article {
  
  @ROUTE()
  async getArticles(req, res) {
    try {
      // find articles will be called 3 times, with 5 sec waiting on each failure
      let articles = await Micro.attampt(
        (curr: number) => articles.find({ ... }),
        { tries: 3, interval: 5000 }
      );
    } catch (e) {
      // after all attempts failed
    }
  }
}

attempt can have a canceler if we want to set timeout for each request

@SERVICE()
class Article {
  
  @ROUTE()
  async getArticles(req, res) {
    try {
      // find articles will be called 3 times, with 5 sec waiting on each failure or cancel on timeout
      let articles = await Micro.attampt(
        (curr: number) => articles.find({ ... }),
        (promise) => {
          // called on current try timeout
          // terminate promise some how
        },
        // setting timeout option
        { tries: 3, interval: 5000, timeout: 5000 }
      );
    } catch (e) {
      // after all attempts failed or canceled on timeout
    }
  }
}

Lifecycle & Events Methods

PMS will try to call some service methods in specific time or action if they were already defined in our service.

onInit

When defined, will be called once our service is instantiated but nothing else, this method is useful when we need to connect to a databese or to make some async operations before start listening one events or http requests.

It can return a promise or nothing.

import { SERVICE, ServiceEvents } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher implements ServiceEvents {

  async onInit() {
    // connect to a databese
  }
}

onReady

This method is called once all our listeners are ready.

import { SERVICE, ServiceEvents } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher implements ServiceEvents {

  onReay() {}
}

onExit

Called once our service is stopped when calling Micro.exit() or when any of termination signals are triggerred SIGTERM, SIGINT, SIGHUP,

Exit code with the signal are passed as arguments.

import { SERVICE, ServiceEvents } from '@pestras/microservice';

@SERVICE({ workers: 4 })
class Publisher implements ServiceEvents {

  onExit(code: number, signal: NodeJS.Signals) {
    // disconnecting from the databese
  }
}

OnLog

PMS has a built in lightweight logger that logs everything to the console.

In order to change that behavior we can define onLog event method in our service and PMS will detect that method and will transfer all logs to it, besides enabling transferLog options in service config.

import { SERVICE, SUBJECT, Micro, ServiceEvents } from '@pestras/microservice';

@SERVICE({
  version: 1
  transferLog: process.env.NODE_ENV === 'production'
})
class Test implements ServiceEvents {

  onLog(level: LOGLEVEL, msg: any, extra: any) {
    // what ever you code
  }

  @SUBJECT({ subject: 'newArticle' })
  newArticle() {
    try {

    } catch (e) {
      Micro.logger.error('some error', e);
    }
  }
}

onHealthcheck

An event triggered for docker swarm healthcheck.

@SERVICE()
class Publisher implements ServiceEvents {

  // http: GET /healthcheck
  async onHealthcheck(res: Response) {
    // check for database connection
    if (dbConnected) res.status(200).end();
    else res.status(500).end()
  }
}

onReadycheck

An event triggered for kubernetes ready check.

@SERVICE()
class Publisher implements ServiceEvents {

  // http: GET /readiness
  async onReadycheck(res: Response) {
    // check for database connection
    if (dbConnected) res.status(200).end();
    else res.status(500).end()
  }
}

onLivecheck

An event triggered for kubernetes live check.

@SERVICE()
class Publisher implements ServiceEvents {

  // http: GET /liveness
  async onLivecheck(res: Response) {
    // check for database connection
    if (dbConnected) res.status(200).end();
    else res.status(500).end()
  }
}

onRequest

Called whenever a new http request is received, passing the Request and Response instances as arguments, it can return a promise or nothing;

@SERVICE()
class Publisher implements ServiceEvents {

  async onRequest(req: Request, res: Response) { }
}

This event method is called before authorizing the request or even before checking if there is a matched route or not.

on404

Called whenever http request has no route handler found.

@SERVICE()
class Publisher implements ServiceEvents {

  on404(req: Request, res: Response) {

  }
}

When implemented response should be implemented as well

onError

Called whenever an error accured when handling an http request, passing the Request and Response instances and the error as arguments.

@SERVICE({ workers: 4 })
class Publisher implements ServiceEvents {

  onError(req: Request, res: Response, err: any) { }
}

onUnhandledRejection

Defining this handler will cancel exitOnUnhandledRejection option in service config, so you need to exit manually if it needs to be.

@SERVICE({ workers: 4 })
class Publisher implements ServiceEvents {

  onUnhandledRejection(reason: any, p: Promise<any>) {
    // do somethig with the error and then maybe exit
    // calling Micro.exit() will trigger onExit EventHandler
    Micro.exit(1);
  }
}

onUnhandledException

Defining this handler will cancel exitOnUnhandledException option in service config, so you need to exit manually if it needs to be.

@SERVICE({ workers: 4 })
class Publisher implements ServiceEvents {

  onUnhandledException(err: any) {
    // do somethig with the error and then maybe exit
    // calling Micro.exit() will trigger onExit EventHandler
    Micro.exit(1);
  }
}

Health Check

For health check in Dockerfile or docker-compose

HEALTHCHECK --interval=1m30s --timeout=2s --start_period=10s CMD node ./node_modules/@pestras/microservice/hc.js /articles/v0 3000

healthcheck:
  test: ["CMD", "node", "./node_modules/@pestras/microservice/hc.js", "/articles/v0", "3000"]
  interval: 1m30s
  timeout: 10s
  retries: 3
  start_period: 40s

Root path is required as the first parameter, while port defaults to 3000.

Thank you