N2O

INTRO

The n2o defines the way you create, configure and run arbitrary applications and protocols inside some hosts, into which N2O can be injected, such as cowboy and emqttd. Each application can spawn its instance in its own way, like web pages spawn WebSocket connections, workflow engines spawn business proceseses and chat applications spawns roster and chatroom processes. With N2O everything is managed by protocols.

N2O shipped to work in two modes: 1) inside cowboy processes, implemented in n2o_stream 2) inside n2o_vnode workers. In second case MQTT server is used for transfering message between clients and server workers. In first case no more processes are being introduced except clients. You can create you own configuration of N2O processing loop.

The N2O itself a an enbeddable protocol loop in n2o_proto. However besides that it handles cache and session ETS tables along with flexible n2o_async processes with no ownership restriction. It also introduces logging approach, AES/CBC—128 pickling and BERT/JSON formatter.

RECORDS

Listing 1. Erlang/OTP records
#ok { data = [] :: term() }. #error { data = [] :: term() }.
Listing 2. N2O Protocol
#reply { data = [] :: [] | { Formatter :: atom(), binary() }, req = [] :: [] | term(), state = [] :: [] | term() }. #unknown { data = [] :: [] | binary(), req = [] :: [] | term(), state = [] :: [] | term() }.
Listing 3. N2O State
#cx { session = [] :: [] | binary(), formatter = bert :: bert | json, actions = [] :: list(tuple()), state = [] :: [] | term(), module = [] :: [] | atom(), lang = [] :: [] | atom(), path = [] :: [] | binary(), node = [] :: [] | atom(), pid = [] :: [] | pid(), vsn = [] :: [] | integer() }).

PROTOCOLS

While all application protocols in the system are desired to be placed in the single effectful environment or same error handling path, n2o defines single protocol loop for all applications in its federation of protocols.

In core bundle n2o is shipped with NITRO and FTP protocols which allows you to create real-time web applications with binary-based protocols, priving also robust and performant upload client and file transfer protocol. For bulding web based NITRO applications you need to include nitro dependency.

info(term(),term(),#cx{}) -> #reply{} | #unknown{}.

The info/2 is a N2O protocol callback that will be called on each incoming request. N2O code should be embedded into applications host: MQTT (as ring of MQTT clients), or HTTP server or WebSocket, or raw TCP.

RPC MQTT

N2O provides RPC over MQ mechanism for MQTT devices. N2O spawn a set of n2o_vnode workers as n2o_async processes that listen to events topic. Response are sent to actions topic, which is subscribed automatically on MQTT session init.

Listing 4. MQTT RPC Topics
actions/:vsn/:module/:client events/:vsn/:node/:module/:client

RPC WebSocket

In pure WebSocket case N2O implements n2o_proto as cowboy module supporting binary and text messages.

Listing 5. Cowboy stream protocol
#binary { data :: binary() }. #text { data :: binary() }.

EXAMPLE

Here is little example of overriding INIT protocol message from NITRO protocol and generate standart token stored in KVS.

Listing 6. Custom INIT Protocol
-module(custom_init). -compile(export_all). info({init, <<>>}, Req, State = #cx{session = Session}) -> {'Token', Token} = n2o_auth:gen_token([], Session), #cx{params = Client} = get(context), kvs:put(#'Token'{token = Token, client = Client}), n2o_nitro:info({init, Token}, Req, State); info(Message,Req,State) -> {unknown,Message,Req,State}.

CONFIG

Just put protocol implementation module name to protocol option in sys.config.

[{n2o,[{cache,n2o}, {mq,n2o}, {logging,n2o_io}, {log_modules,n2o}, {log_level,n2o}, {session,n2o_session}, {pickler,n2o_secret}, {protocols,[custom_init,n2o_ftp,n2o_nitro]}, {timer,{0,10,0}}]}].

N2O is the facade of the following services: cache, mq, message formating, loging, sessions, pickling and protocol loops. The other part of N2O is n2o_async module for spawning supervised application processes tha use N2O API. In this simple configuration you may set any implementation to any service.

CACHE

Cache is fast expirable memory store. Just put values onto keys using these functions and system timer will clear expired entries eventually. You can select caching module implementation by seting cache n2o parameter to module name. Default n2o cache implementation just turns each ets store into expirable.

cache(Tab, Key, Value, Till) -> term().

Sets a Value with a given TTL.

cache(Tab, Key) -> term().

Gets a Value.

PUBSUB

The minimal requirement for any framework is to pub/sub API. N2O provides selectable API through mq environment parameter.

subscribe(Client, Topic, Options) -> term().

Subscribe an absctract client to a transient topic. In particular implementation the semantics could differ. In MQTT you can subscribe offline/online clients to any persistent topics.

unsubscribe(Client, Topic, Options) -> term().

Unsubscribe an abstract client from a transient topic. In MQTT we remove the subscription from persistent database.

publish(Topic, Message, Options) -> term().

Publish a message to a topic. In MQTT if clients are offline they will receive offline messages from the inflight srotarge once they become online.

FORMAT

You specify formatter in the #reply/3 return from procotol. N2O is shipped with two formatters: {bert,_} and {json,_}. You can use {binary,_} for unformatted messages. For unhandled messages protocol should return in #reply.data {unknown,_}.

format(Message) -> binary().

Here is example of formatter and its usage in info/3 protocol implementation.

-module(ldap). -inlcude("LDAP.hrl"). -export([format/1]). format(Term) -> element(2,'LDAP':encode(element(1,Term),Term)). info(#'LDAPMessage', R, S) -> {reply, {ldap, M}, R, S}. info(M, R, S) -> {unknown, M, R, S}. > ldap:format(#'LDAPMessage'{messageID=2,protocolOp={unbindRequest,3}})). <<48,5,2,1,2,66,0>>

LOG

First you need specify global module in sys.config, where functions log_level and log_modules are placed. See options in config with with same names as functions.

Then implement these function in way of returning the list modules you want to trace, and global log level for them.

Listing 7. Log Framework
log_modules() -> [n2o,n2o_async,n2o_proto]. log_level() -> info.

In your code you should use following trace functions which are the same as callback API for n2o logging environment variable.

error(Module, Format, Args) -> ok | skip.

info(Module, Format, Args) -> ok | skip.

warning(Module, Format, Args) -> ok | skip.

SESSION

Sessions are stored in cookies table and indexed by security token which is usually a password based token. All session variables from all users are stored in this table. each user see only its variables indexed by his token. Sessions like a cache are expirable. Technically, N2O sessions are the server controlling mechanism of JavaScript cookies.

session(Key, Value) -> term().

Sets a Value into ETS table cookies for a token from #cx.session which is set there earlier from INIT message or MQTT headers, before entering the top level N2O loop.

Listing 8. Sessions
1> rr(n2o). [bin,client,cx,direct,ev,flush,ftp,ftpack,handler, mqtt_client,mqtt_message,pickle,server] 2> put(context,#cx{session=10}). undefined 3> n2o:session(user,maxim). maxim 4> ets:tab2list(cookies). [{{10,user}, <<"/">>, {1504,977449,476430}, {{2017,9,9},{20,32,29}}, maxim}]

session(Key) -> term().

Gets a Value by any Key.

PICKLE

Call this function for changeable at runtime term pickling.

pickle(term()) -> binary().

depickle(binary()) -> term().

This module may refer to: io, ets, n2o_asynrc, n2o_vnode, n2o_proto.