ChannelContext#clientId

Unique persistence client ID.

server.clientIds.get(node.clientId)

Type: string.

ChannelContext#data

Open structure to save some data between different steps of processing.

server.type('RENAME', {
  access (ctx, action, meta) {
    ctx.data.user = findUser(ctx.userId)
    return ctx.data.user.hasAccess(action.projectId)
  }
  process (ctx, action, meta) {
    return ctx.data.user.rename(action.projectId, action.name)
  }
})

Type: Data.

ChannelContext#headers

Client’s headers.

ctx.sendBack({
  type: 'error',
  message: I18n[ctx.headers.locale || 'en'].error
})

Type: Headers.

ChannelContext#isServer

Was action created by Logux server.

access: (ctx, action, meta) => ctx.isServer

Type: boolean.

ChannelContext#nodeId

Unique node ID.

server.nodeIds.get(node.nodeId)

Type: string.

ChannelContext#params

Parsed variable parts of channel pattern.

server.channel('user/:id', {
  access (ctx, action, meta) {
    action.channel //=> user/10
    ctx.params //=> { id: '10' }
  }
})
server.channel(/post/(\d+)/, {
  access (ctx, action, meta) {
    action.channel //=> post/10
    ctx.params //=> ['post/10', '10']
  }
})

Type: ChannelParams.

ChannelContext#server

Logux server

Type: Server.

ChannelContext#subprotocol

Action creator application subprotocol version.

Type: number.

ChannelContext#userId

User ID taken node ID.

async access (ctx, action, meta) {
  const user = await db.getUser(ctx.userId)
  return user.admin
}

Type: string.

ChannelContext#drain()

Wait until the client will confirm all actions, which were sent to it.

Use it to send a long history without loading it all into the memory: the client’s speed will limit how fast you read the database.

while (await ctx.drain()) {
  let page = await cursor.next(100)
  if (!page.length) break
  ctx.sendBack(page.map(i => i.action))
}

Returns Promise<boolean>. Promise with false if the client was disconnected.

ChannelContext#sendBack(action, meta?)

Send action back to the client.

ctx.sendBack({ type: 'login/success', token })

An array of actions will be sent in a single message. Use it to send a big history page by page instead of a message per action.

ctx.sendBack(page.map(i => i.action))

Every action in the array can have own meta as [action, meta].

Action will not be processed by server’s callbacks from Server#type.

Returns Promise. Promise until action was added to the server log.

Context#clientId

Unique persistence client ID.

server.clientIds.get(node.clientId)

Type: string.

Context#data

Open structure to save some data between different steps of processing.

server.type('RENAME', {
  access (ctx, action, meta) {
    ctx.data.user = findUser(ctx.userId)
    return ctx.data.user.hasAccess(action.projectId)
  }
  process (ctx, action, meta) {
    return ctx.data.user.rename(action.projectId, action.name)
  }
})

Type: Data.

Context#headers

Client’s headers.

ctx.sendBack({
  type: 'error',
  message: I18n[ctx.headers.locale || 'en'].error
})

Type: Headers.

Context#isServer

Was action created by Logux server.

access: (ctx, action, meta) => ctx.isServer

Type: boolean.

Context#nodeId

Unique node ID.

server.nodeIds.get(node.nodeId)

Type: string.

Context#server

Logux server

Type: Server.

Context#subprotocol

Action creator application subprotocol version.

Type: number.

Context#userId

User ID taken node ID.

async access (ctx, action, meta) {
  const user = await db.getUser(ctx.userId)
  return user.admin
}

Type: string.

Context#drain()

Wait until the client will confirm all actions, which were sent to it.

Use it to send a long history without loading it all into the memory: the client’s speed will limit how fast you read the database.

while (await ctx.drain()) {
  let page = await cursor.next(100)
  if (!page.length) break
  ctx.sendBack(page.map(i => i.action))
}

Returns Promise<boolean>. Promise with false if the client was disconnected.

Context#sendBack(action, meta?)

Send action back to the client.

ctx.sendBack({ type: 'login/success', token })

An array of actions will be sent in a single message. Use it to send a big history page by page instead of a message per action.

ctx.sendBack(page.map(i => i.action))

Every action in the array can have own meta as [action, meta].

Action will not be processed by server’s callbacks from Server#type.

Returns Promise. Promise until action was added to the server log.

ResponseError#name

Type: "ResponseError".

ResponseError#statusCode

Type: number.

Server.loadOptions(process, defaults)

Load options from command-line arguments and/or environment.

const server = new Server(Server.loadOptions(process, {
  minSubprotocol: 1,
  subprotocol: 1,
  root: import.meta.dirname,
  port: 31337
}))

Returns ServerOptions. Parsed options object.

Server#clientIds

Connected client by client ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient>.

Server#connected

Connected clients.

for (let client of server.connected.values()) {
  console.log(client.remoteAddress)
}

Type: Map<string,ServerClient>.

Server#env

Production or development mode.

if (server.env === 'development') {
  logDebugData()
}

Type: "development" | "production".

Server#log

Server actions log.

server.log.each(finder)

Type: Log.

Server#logger

Console for custom log records. It uses pino API.

server.on('connected', client => {
  server.logger.info(
    { domain: client.httpHeaders.domain },
    'Client domain'
  )
})

Type: { debug: LogFn, error: LogFn, fatal: LogFn, info: LogFn, warn: LogFn }.

Server#nodeId

Server unique ID.

console.log('Error was raised on ' + server.nodeId)

Type: string.

Server#nodeIds

Connected client by node ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient>.

Server#options

Server options.

console.log('Server options', server.options.subprotocol)

Type: ServerOptions.

Server#subscribers

Clients subscribed to some channel.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: { }.

Server#userIds

Connected client by user ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient[]>.

Server#addClient(connection)

Add new client for server. You should call this method manually mostly for test purposes.

server.addClient(test.right)

Returns number. Client ID.

Server#auth(authenticator)

Set authenticate function. It will receive client credentials and node ID. It should return a Promise with true or false.

server.auth(async ({ userId, cookie }) => {
  const user = await findUserByToken(cookie.token)
  return !!user && userId === user.id
})

Server#autoloadModules(files?)

Load module creators and apply to the server. By default, it will load files from modules/*.

await server.autoloadModules()

Returns Promise.

Server#channel(pattern, callbacks, options?)

Define the channel.

server.channel('user/:id', {
  access (ctx, action, meta) {
    return ctx.params.id === ctx.userId
  }
  filter (ctx, action, meta) {
    return (otherCtx, otherAction, otherMeta) => {
      return !action.hidden
    }
  }
  async load (ctx, action, meta) {
    const user = await db.loadUser(ctx.params.id)
    ctx.sendBack({ type: 'USER_NAME', name: user.name })
  }
})

Server#debugError(error)

Send runtime error stacktrace to all clients.

process.on('uncaughtException', e => {
  server.debugError(e)
})

Server#destroy()

Stop server and unbind all listeners.

afterEach(() => {
  testServer.destroy()
})

Returns Promise. Promise when all listeners will be removed.

Server#drain(clientId)

Wait until the client will confirm all actions, which were sent to it.

Use it to send a long history without loading it all into the memory: the client’s speed will limit how fast you read the database.

while (await server.drain(clientId)) {
  let page = await cursor.next(100)
  if (!page.length) break
  server.log.add(page.map(i => [i.action, { clients: [clientId] }]))
}

Returns Promise<boolean>. Promise with false if the client was disconnected.

Server#handleClient(ws, req)

Handle WebSocket connection explicitly

This is a low-level method allowing to integrate Logux server with an existing server

fastify.get('/', { websocket: true }, (socket, req) => {
  loguxServer.handleClient(socket, req)
})

Server#http(method, url, listener)

Add non-WebSocket HTTP request processor.

server.http('GET', '/auth', (req, res) => {
  let token = signIn(req)
  if (token) {
    res.setHeader('Set-Cookie', `token=${token}; Secure; HttpOnly`)
    res.end()
  } else {
    res.statusCode = 400
    res.end('Wrong user or password')
  }
})

Server#listen()

Start WebSocket server and listen for clients.

Returns Promise. When the server has been bound.

Server#on(event, listener)

Returns Unsubscribe.

Server#otherChannel(callbacks)

Set callbacks for unknown channel subscription.

server.otherChannel({
  async access (ctx, action, meta) {
    const res = await phpBackend.checkChannel(ctx.params[0], ctx.userId)
    if (res.code === 404) {
      this.wrongChannel(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
})

Server#otherType(callbacks)

Define callbacks for actions, which type was not defined by any Server#type. Useful for proxy or some hacks.

Without this settings, server will call Server#unknownType on unknown type.

server.otherType(
  async access (ctx, action, meta) {
    const response = await phpBackend.checkByHTTP(action, meta)
    if (response.code === 404) {
      this.unknownType(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
  async process (ctx, action, meta) {
    return await phpBackend.sendHTTP(action, meta)
  }
})

Server#process(action, meta?)

Add new action to the server and return the Promise until it will be resend to clients and processed.

Returns Promise<ServerMeta>. Promise until new action will be resend to clients and processed.

Server#sendAction(action, meta)

Send action, received by other server, to all clients of current server. This method is for multi-server configuration only.

server.on('add', (action, meta) => {
  if (meta.server === server.nodeId) {
    sendToOtherServers(action, meta)
  }
})
onReceivingFromOtherServer((action, meta) => {
  server.sendAction(action, meta)
})

Returns void | Promise.

Server#sendOnConnect(loader)

Change a way how server loads actions history for the client.

server.sendOnConnect(async (ctx, lastSynced) => {
  return db.loadActions({ user: ctx.userId, after: lastSynced })
})

Actions should be returned from the newest one to the oldest one. The list will be split into messages by the syncBatch option, but it is loaded into the memory as a whole. For a big history send it by pages with Context#sendBack() and Context#drain() instead.

Set meta.added to let the client ask only for newer actions after the reconnect: the biggest added will be sent as the sync position.

Server#subscribe(nodeId, channel)

Send logux/subscribed if client was not already subscribed.

server.subscribe(ctx.nodeId, `users/${loaded}`)

Server#type(actionCreator, callbacks, options?)

Server#undo(action, meta, reason?, extra?)

Undo action from client.

if (couldNotFixConflict(action, meta)) {
  server.undo(action, meta)
}

Returns Promise. When action was saved to the log.

Server#unknownType(action, meta)

If you receive action with unknown type, this method will mark this action with error status and undo it on the clients.

If you didn’t set Server#otherType, Logux will call it automatically.

server.otherType({
  access (ctx, action, meta) {
    if (action.type.startsWith('myapp/')) {
      return proxy.access(action, meta)
    } else {
      server.unknownType(action, meta)
    }
  }
})

Server#wrongChannel(action, meta)

Report that client try to subscribe for unknown channel.

Logux call it automatically, if you will not set Server#otherChannel.

server.otherChannel({
  async access (ctx, action, meta) {
    const res = phpBackend.checkChannel(params[0], ctx.userId)
    if (res.code === 404) {
      this.wrongChannel(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
})

ServerClient#app

Server, which received client.

Type: BaseServer.

ServerClient#clientId

Unique persistence machine ID. It will be undefined before correct authentication.

Type: string.

ServerClient#connection

The Logux wrapper to WebSocket connection.

console.log(client.connection.ws.upgradeReq.headers)

Type: ServerConnection.

ServerClient#data

Open structure to save data for the whole connection.

server.auth(async ({ client, userId, token }) => {
  let session = await findSession(token)
  if (!session) return false
  client.data.sessionId = session.id
  return true
})
server.on('disconnected', client => {
  touchSession(client.data.sessionId)
})

Type: ClientData.

ServerClient#httpHeaders

HTTP headers of WS connection.

client.httpHeaders['User-Agent']

Type: { }.

ServerClient#key

Client number used as app.connected key.

function stillConnected (client) {
  return app.connected.has(client.key)
}

Type: string.

ServerClient#node

Node instance to synchronize logs.

if (client.node.state === 'synchronized')

Type: ServerNode.

ServerClient#nodeId

Unique node ID. It will be undefined before correct authentication.

Type: string.

ServerClient#processing

Does server process some action from client.

console.log('Clients in processing:', clients.map(i => i.processing))

Type: boolean.

ServerClient#remoteAddress

Client IP address.

const clientCity = detectLocation(client.remoteAddress)

Type: string.

ServerClient#userId

User ID. It will be filled from client’s node ID. It will be undefined before correct authentication.

Type: string.

ServerClient#destroy()

Disconnect client.

ServerClient#drain()

Wait until the client will confirm all sent actions.

Returns Promise<boolean>. Promise with false if the client was disconnected.

TestClient#clientId

Client’s ID.

let client = new TestClient(server, '10')
client.clientId //=> '10:1'

Type: string.

TestClient#log

Client’s log with extra methods to check actions inside.

console.log(client.log.entries())

Type: TestLog.

TestClient#node

Logux node.

Type: ClientNode.

TestClient#nodeId

Client’s node ID.

let client = new TestClient(server, '10')
client.nodeId //=> '10:1:1'

Type: string.

TestClient#pair

Connection channel between client and server to track sent messages.

console.log(client.pair.leftSent)

Type: TestPair.

TestClient#userId

User ID.

let client = new TestClient(server, '10')
client.userId //=> '10'

Type: string.

TestClient#collect(test)

Collect actions added by server and other clients during the test call.

let answers = await client.collect(async () => {
  client.log.add({ type: 'pay' })
  await delay(10)
})
expect(actions).toEqual([{ type: 'paid' }])

Returns Promise<Action[]>. Promise with all received actions

TestClient#connect(opts?)

Connect to test server.

let client = new TestClient(server, '10')
await client.connect()

Returns Promise. Promise until the authorization.

TestClient#disconnect()

Disconnect from test server.

await client.disconnect()

Returns Promise. Promise until connection close.

TestClient#process(action, meta?)

Send action to the sever and collect all response actions.

await client.process({ type: 'posts/add', post })
let posts = await client.subscribe('posts')
expect(posts).toHaveLength(1)

Returns Promise<Action[]>. Promise until logux/processed answer.

TestClient#received(test)

Collect actions received from server during the test call.

let answers = await client1.received(async () => {
  await client2.process({ type: 'resend' })
})
expect(actions).toEqual([{ type: 'local' }])

Returns Promise<Action[]>. Promise with all received actions

TestClient#subscribe(channel, filter?, since?)

Subscribe to the channel and collect all actions during the subscription.

let posts = await client.subscribe('posts')
expect(posts).toEqual([
  { type: 'posts/add', post }
])

Returns Promise<Action[]>. Promise with all actions from the server.

TestClient#unsubscribe(channel, filter?)

Unsubscribe client from the channel.

await client.unsubscribe('posts')

Returns Promise<Action[]>. Promise until server will remove client from subscribers.

TestLog#nodeId

Unique node ID. It is used in action IDs.

Type: string.

TestLog#store

Log store.

Type: LogStore.

TestLog#actions()

Return all action (without metadata) inside log, sorted by created time.

This shortcut works only with MemoryStore.

expect(log.action).toEqual([
  { type: 'A' }
])

Returns Action[].

TestLog#add(action, meta?)

Add action to log.

It will set id, time (if they was missed) and added property to meta and call all listeners.

removeButton.addEventListener('click', () => {
  log.add({ type: 'users:remove', user: id })
})

Returns Promise<false | LogMeta>. Promise with meta if action was added to log or false if action was already in log.

TestLog#addReason(reasons, criteria?)

Add reason tags to metadata of actions, which are already in the log. Reasons, which action already has, will not be duplicated.

// The action still owns these cells, so it should keep their reasons
log.addReason('last-value', { id: meta.id })

Returns Promise. Promise when adding will be finished.

TestLog#byId(id)

Does log already has action with this ID.

if (action.type === 'logux/undo') {
  const [undidAction, undidMeta] = await log.byId(action.id)
  log.changeMeta(meta.id, { reasons: undidMeta.reasons })
}

Returns Promise<[null, null] | [Action, LogMeta]>. Promise with array of action and metadata.

TestLog#changeMeta(id, diff)

Change action metadata. You will remove action by setting reasons: [].

await process(action)
log.changeMeta(action, { status: 'processed' })

Returns Promise<boolean>. Promise with true if metadata was changed or false on unknown ID.

TestLog#each(opts, callback)

Returns Promise.

TestLog#entries()

Return all entries (with metadata) inside log, sorted by created time.

This shortcut works only with MemoryStore.

expect(log.action).toEqual([
  [{ type: 'A' }, { id: '1 test1 0', time: 1, added: 1, reasons: ['t'] }]
])

Returns [Action, ServerMeta][].

TestLog#generateId()

Generate next unique action ID.

const id = log.generateId()

Returns string. Unique ID for action.

TestLog#keepActions()

Keep actions without meta.reasons in the log by setting test reason during adding to the log.

log.keepActions()
log.add({ type: 'test' })
log.actions() //=> [{ type: 'test' }]

TestLog#now()

Current time of this node. Log uses it for meta.time and action ID.

Redefine it to use a custom clock, for instance, in tests.

log.now = () => fakeTime

Returns number. Milliseconds since UNIX epoch.

TestLog#on(event, listener)

Subscribe for log events. It implements nanoevents API. Supported events:

• preadd: when somebody try to add action to log. It fires before ID check. The best place to add reason.
• add: when new action was added to log.
• clean: when action was cleaned from store.
• batch: when actions from a single Log#add() call were added.

Note, that Log#type() will work faster than on event with if.

log.on('preadd', (action, meta) => {
  if (action.type === 'beep') {
    meta.reasons.push('test')
  }
})

Returns Unsubscribe. Unbind listener from event.

TestLog#removeReason(reasons, criteria?)

Remove reason tags from actions’ metadata and remove actions without reasons from log.

onSync(lastSent) {
  log.removeReason('unsynchronized', { maxAdded: lastSent })
}

Returns Promise. Promise when cleaning will be finished.

TestLog#type(type, listener, opts?)

Add listener for adding action with specific type. Works faster than on('add', cb) with if.

Setting opts.id will filter events ponly from actions with specific action.id.

const unbind = log.type('beep', (action, meta) => {
  beep()
})
function disableBeeps () {
  unbind()
}

Returns Unsubscribe. Unbind listener from event.

TestPair#delay

Delay for connection and send events to emulate real connection latency.

Type: number.

TestPair#left

First connection. Will be connected to right one after connect().

new ClientNode('client, log1, pair.left)

Type: LocalConnection.

TestPair#leftEvents

Emitted events from left connection.

await pair.left.connect()
pair.leftEvents //=> [['connect']]

Type: string[][].

TestPair#leftNode

Node instance used in this test, connected with left.

function createTest () {
  test = new TestPair()
  test.leftNode = ClientNode('client', log, test.left)
  return test
}

Type: BaseNode.

TestPair#leftSent

Sent messages from left connection.

await pair.left.send(msg)
pair.leftSent //=> [msg]

Type: Message[].

TestPair#right

Second connection. Will be connected to right one after connect().

new ServerNode('server, log2, pair.right)

Type: LocalConnection.

TestPair#rightEvents

Emitted events from right connection.

await pair.right.connect()
pair.rightEvents //=> [['connect']]

Type: string[][].

TestPair#rightNode

Node instance used in this test, connected with right.

function createTest () {
  test = new TestPair()
  test.rightNode = ServerNode('client', log, test.right)
  return test
}

Type: BaseNode.

TestPair#rightSent

Sent messages from right connection.

await pair.right.send(msg)
pair.rightSent //=> [msg]

Type: Message[].

TestPair#clear()

Clear all connections events and messages to test only last events.

await client.connection.connect()
pair.clear() // Remove all connecting messages
await client.log.add({ type: 'a' })
expect(pair.leftSent).toEqual([
  ['sync', …]
])

TestPair#wait(receiver?)

Return Promise until next event.

pair.left.send(['test'])
await pair.wait('left')
pair.leftSend //=> [['test']]

Returns Promise<TestPair>. Promise until next event.

TestServer#clientIds

Connected client by client ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient>.

TestServer#connected

Connected clients.

for (let client of server.connected.values()) {
  console.log(client.remoteAddress)
}

Type: Map<string,ServerClient>.

TestServer#env

Production or development mode.

if (server.env === 'development') {
  logDebugData()
}

Type: "development" | "production".

TestServer#fetch

fetch() compatible API to test HTTP endpoints.

server.http('GET', '/version', (req, res) => {
  res.end('1.0.0')
})
let res = await server.fetch()
expect(await res.text()).toEqual('1.0.0')

Type: (input: URL | RequestInfo, init?: RequestInit) => Promise<Response>.

TestServer#log

Server actions log, with methods to check actions inside.

server.log.actions() //=> […]

Type: TestLog.

TestServer#logger

Console for custom log records. It uses pino API.

server.on('connected', client => {
  server.logger.info(
    { domain: client.httpHeaders.domain },
    'Client domain'
  )
})

Type: { debug: LogFn, error: LogFn, fatal: LogFn, info: LogFn, warn: LogFn }.

TestServer#nodeId

Server unique ID.

console.log('Error was raised on ' + server.nodeId)

Type: string.

TestServer#nodeIds

Connected client by node ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient>.

TestServer#options

Server options.

console.log('Server options', server.options.subprotocol)

Type: BaseServerOptions.

TestServer#subscribers

Clients subscribed to some channel.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: { }.

TestServer#time

Time replacement without variable parts like current timestamp.

Type: TestTime.

TestServer#userIds

Connected client by user ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient[]>.

TestServer#addClient(connection)

Add new client for server. You should call this method manually mostly for test purposes.

server.addClient(test.right)

Returns number. Client ID.

TestServer#auth(authenticator)

Set authenticate function. It will receive client credentials and node ID. It should return a Promise with true or false.

server.auth(async ({ userId, cookie }) => {
  const user = await findUserByToken(cookie.token)
  return !!user && userId === user.id
})

TestServer#channel(pattern, callbacks, options?)

Define the channel.

server.channel('user/:id', {
  access (ctx, action, meta) {
    return ctx.params.id === ctx.userId
  }
  filter (ctx, action, meta) {
    return (otherCtx, otherAction, otherMeta) => {
      return !action.hidden
    }
  }
  async load (ctx, action, meta) {
    const user = await db.loadUser(ctx.params.id)
    ctx.sendBack({ type: 'USER_NAME', name: user.name })
  }
})

TestServer#connect(userId, opts?)

Create and connect client.

server = new TestServer()
let client = await server.connect('10')

Returns Promise<TestClient>. Promise with new client.

TestServer#debugError(error)

Send runtime error stacktrace to all clients.

process.on('uncaughtException', e => {
  server.debugError(e)
})

TestServer#destroy()

Stop server and unbind all listeners.

afterEach(() => {
  testServer.destroy()
})

Returns Promise. Promise when all listeners will be removed.

TestServer#drain(clientId)

Wait until the client will confirm all actions, which were sent to it.

Use it to send a long history without loading it all into the memory: the client’s speed will limit how fast you read the database.

while (await server.drain(clientId)) {
  let page = await cursor.next(100)
  if (!page.length) break
  server.log.add(page.map(i => [i.action, { clients: [clientId] }]))
}

Returns Promise<boolean>. Promise with false if the client was disconnected.

TestServer#expectDenied(test)

Call callback and throw an error if there was no Action was denied during callback.

await server.expectDenied(async () => {
  client.subscribe('secrets')
})

Returns Promise.

TestServer#expectError(text, test)

Call callback and throw an error if there was no error during server processing.

Returns Promise.

TestServer#expectUndo(reason, test)

Call callback and throw an error if there was no logux/undo in return with specific reason.

await server.expectUndo('notFound', async () => {
  client.subscribe('projects/nothing')
})

Returns Promise.

TestServer#expectWrongCredentials(userId, opts?)

Try to connect client and throw an error is client didn’t received Wrong Cregentials message from the server.

server = new TestServer()
await server.expectWrongCredentials('10')

Returns Promise. Promise until check.

TestServer#handleClient(ws, req)

Handle WebSocket connection explicitly

This is a low-level method allowing to integrate Logux server with an existing server

fastify.get('/', { websocket: true }, (socket, req) => {
  loguxServer.handleClient(socket, req)
})

TestServer#http(method, url, listener)

Add non-WebSocket HTTP request processor.

server.http('GET', '/auth', (req, res) => {
  let token = signIn(req)
  if (token) {
    res.setHeader('Set-Cookie', `token=${token}; Secure; HttpOnly`)
    res.end()
  } else {
    res.statusCode = 400
    res.end('Wrong user or password')
  }
})

TestServer#listen()

Start WebSocket server and listen for clients.

Returns Promise. When the server has been bound.

TestServer#on(event, listener)

Returns Unsubscribe.

TestServer#otherChannel(callbacks)

Set callbacks for unknown channel subscription.

server.otherChannel({
  async access (ctx, action, meta) {
    const res = await phpBackend.checkChannel(ctx.params[0], ctx.userId)
    if (res.code === 404) {
      this.wrongChannel(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
})

TestServer#otherType(callbacks)

Define callbacks for actions, which type was not defined by any Server#type. Useful for proxy or some hacks.

Without this settings, server will call Server#unknownType on unknown type.

server.otherType(
  async access (ctx, action, meta) {
    const response = await phpBackend.checkByHTTP(action, meta)
    if (response.code === 404) {
      this.unknownType(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
  async process (ctx, action, meta) {
    return await phpBackend.sendHTTP(action, meta)
  }
})

TestServer#process(action, meta?)

Add new action to the server and return the Promise until it will be resend to clients and processed.

Returns Promise<ServerMeta>. Promise until new action will be resend to clients and processed.

TestServer#sendAction(action, meta)

Send action, received by other server, to all clients of current server. This method is for multi-server configuration only.

server.on('add', (action, meta) => {
  if (meta.server === server.nodeId) {
    sendToOtherServers(action, meta)
  }
})
onReceivingFromOtherServer((action, meta) => {
  server.sendAction(action, meta)
})

Returns void | Promise.

TestServer#sendOnConnect(loader)

Change a way how server loads actions history for the client.

server.sendOnConnect(async (ctx, lastSynced) => {
  return db.loadActions({ user: ctx.userId, after: lastSynced })
})

Actions should be returned from the newest one to the oldest one. The list will be split into messages by the syncBatch option, but it is loaded into the memory as a whole. For a big history send it by pages with Context#sendBack() and Context#drain() instead.

Set meta.added to let the client ask only for newer actions after the reconnect: the biggest added will be sent as the sync position.

TestServer#subscribe(nodeId, channel)

Send logux/subscribed if client was not already subscribed.

server.subscribe(ctx.nodeId, `users/${loaded}`)

TestServer#type(actionCreator, callbacks, options?)

TestServer#undo(action, meta, reason?, extra?)

Undo action from client.

if (couldNotFixConflict(action, meta)) {
  server.undo(action, meta)
}

Returns Promise. When action was saved to the log.

TestServer#unknownType(action, meta)

If you receive action with unknown type, this method will mark this action with error status and undo it on the clients.

If you didn’t set Server#otherType, Logux will call it automatically.

server.otherType({
  access (ctx, action, meta) {
    if (action.type.startsWith('myapp/')) {
      return proxy.access(action, meta)
    } else {
      server.unknownType(action, meta)
    }
  }
})

TestServer#wrongChannel(action, meta)

Report that client try to subscribe for unknown channel.

Logux call it automatically, if you will not set Server#otherChannel.

server.otherChannel({
  async access (ctx, action, meta) {
    const res = phpBackend.checkChannel(params[0], ctx.userId)
    if (res.code === 404) {
      this.wrongChannel(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
})

TestTime.getLog(opts?)

Shortcut to create time and generate single log. Use it only if you need one log in test.

it('tests log', () => {
  const log = TestTime.getLog()
})

Returns TestLog.

TestTime#lastId

Last letd number in log’s nodeId.

Type: number.

TestTime#nextLog(opts?)

Return next test log in same time.

it('tests 2 logs', () => {
  const time = new TestTime()
  const log1 = time.nextLog()
  const log2 = time.nextLog()
})

Returns TestLog.

BaseNode#authenticated

Did we finish remote node authentication.

Type: boolean.

BaseNode#connected

Is synchronization in process.

node.on('disconnect', () => {
  node.connected //=> false
})

Type: boolean.

BaseNode#connection

Connection used to communicate to remote node.

Type: Connection.

BaseNode#initializing

Promise for node data initial loadiging.

Type: Promise.

BaseNode#lastReceived

Latest remote node’s log added time, which was successfully synchronized. It will be saves in log store.

Type: number.

BaseNode#lastSent

Latest current log added time, which was successfully synchronized. It will be saves in log store.

Type: number.

BaseNode#localNodeId

Unique current machine name.

console.log(node.localNodeId + ' is started')

Type: string.

BaseNode#localProtocol

Used Logux protocol.

if (tool.node.localProtocol !== 1) {
  throw new Error('Unsupported Logux protocol')
}

Type: number.

BaseNode#log

Log for synchronization.

Type: NodeLog.

BaseNode#minProtocol

Minimum version of Logux protocol, which is supported.

console.log(`You need Logux protocol ${node.minProtocol} or higher`)

Type: number.

BaseNode#options

Synchronization options.

Type: NodeOptions.

BaseNode#remoteHeaders

Headers set by remote node. By default, it is an empty object.

let message = I18N_ERRORS[node.remoteHeaders.language || 'en']
node.log.add({ type: 'error', message })

Type: Headers | EmptyHeaders.

BaseNode#remoteNodeId

Unique name of remote machine. It is undefined until nodes handshake.

console.log('Connected to ' + node.remoteNodeId)

Type: string | undefined.

BaseNode#remoteProtocol

Remote node Logux protocol. It is undefined until nodes handshake.

if (node.remoteProtocol >= 5) {
  useNewAPI()
} else {
  useOldAPI()
}

Type: number | undefined.

BaseNode#remoteSubprotocol

Remote node’s application subprotocol version.

It is undefined until nodes handshake. If remote node will not send on handshake its subprotocol, it will be set to 0.

if (node.remoteSubprotocol > 9) {
  useNewAPI()
} else {
  useOldAPI()
}

Type: number | undefined.

BaseNode#state

Current synchronization state.

• disconnected: no connection.
• connecting: connection was started and we wait for node answer.
• sending: new actions was sent, waiting for answer.
• synchronized: all actions was synchronized and we keep connection.

node.on('state', () => {
  if (node.state === 'sending') {
    console.log('Do not close browser')
  }
})

Type: NodeState.

BaseNode#timeFix

Time difference between nodes.

Type: number.

BaseNode#catch(listener)

Disable throwing a error on error message and create error listener.

node.catch(error => {
  console.error(error)
})

Returns Unsubscribe. Unbind listener from event.

BaseNode#destroy()

Shut down the connection and unsubscribe from log events.

connection.on('disconnect', () => {
  server.destroy()
})

BaseNode#on(event, listener)

Returns Unsubscribe.

BaseNode#setLocalHeaders(headers)

Set headers for current node.

if (navigator) {
  node.setLocalHeaders({ language: navigator.language })
}
node.connection.connect()

BaseNode#waitFor(state)

Return Promise until sync will have specific state.

If current state is correct, method will return resolved Promise.

await node.waitFor('synchronized')
console.log('Everything is synchronized')

Returns Promise. Promise until specific state.

Connection#connected

Is connection is enabled.

Type: boolean.

Connection#destroy

Disconnect and unbind all even listeners.

Type: () => void.

Connection#connect()

Start connection. Connection should be in disconnected state from the beginning and start connection only on this method call.

This method could be called again if connection moved to disconnected state.

Returns Promise. Promise until connection will be established.

Connection#disconnect(reason?)

Finish current connection.

Connection#on(event, listener)

Returns Unsubscribe.

Connection#send(message)

Send message to connection.

Log#nodeId

Unique node ID. It is used in action IDs.

Type: string.

Log#store

Log store.

Type: Store.

Log#add(action, meta?)

Add action to log.

It will set id, time (if they was missed) and added property to meta and call all listeners.

removeButton.addEventListener('click', () => {
  log.add({ type: 'users:remove', user: id })
})

Returns Promise<false | LogMeta>. Promise with meta if action was added to log or false if action was already in log.

Log#addReason(reasons, criteria?)

Add reason tags to metadata of actions, which are already in the log. Reasons, which action already has, will not be duplicated.

// The action still owns these cells, so it should keep their reasons
log.addReason('last-value', { id: meta.id })

Returns Promise. Promise when adding will be finished.

Log#byId(id)

Does log already has action with this ID.

if (action.type === 'logux/undo') {
  const [undidAction, undidMeta] = await log.byId(action.id)
  log.changeMeta(meta.id, { reasons: undidMeta.reasons })
}

Returns Promise<[null, null] | [Action, LogMeta]>. Promise with array of action and metadata.

Log#changeMeta(id, diff)

Change action metadata. You will remove action by setting reasons: [].

await process(action)
log.changeMeta(action, { status: 'processed' })

Returns Promise<boolean>. Promise with true if metadata was changed or false on unknown ID.

Log#each(opts, callback)

Returns Promise.

Log#generateId()

Generate next unique action ID.

const id = log.generateId()

Returns string. Unique ID for action.

Log#now()

Current time of this node. Log uses it for meta.time and action ID.

Redefine it to use a custom clock, for instance, in tests.

log.now = () => fakeTime

Returns number. Milliseconds since UNIX epoch.

Log#on(event, listener)

Subscribe for log events. It implements nanoevents API. Supported events:

• preadd: when somebody try to add action to log. It fires before ID check. The best place to add reason.
• add: when new action was added to log.
• clean: when action was cleaned from store.
• batch: when actions from a single Log#add() call were added.

Note, that Log#type() will work faster than on event with if.

log.on('preadd', (action, meta) => {
  if (action.type === 'beep') {
    meta.reasons.push('test')
  }
})

Returns Unsubscribe. Unbind listener from event.

Log#removeReason(reasons, criteria?)

Remove reason tags from actions’ metadata and remove actions without reasons from log.

onSync(lastSent) {
  log.removeReason('unsynchronized', { maxAdded: lastSent })
}

Returns Promise. Promise when cleaning will be finished.

Log#type(type, listener, opts?)

Add listener for adding action with specific type. Works faster than on('add', cb) with if.

Setting opts.id will filter events ponly from actions with specific action.id.

const unbind = log.type('beep', (action, meta) => {
  beep()
})
function disableBeeps () {
  unbind()
}

Returns Unsubscribe. Unbind listener from event.

MemoryStore#entries

Actions in the store.

Type: [Action, ServerMeta][].

MemoryStore#add(action, meta)

Add action to store. Action always will have type property.

Returns Promise<false | ServerMeta>. Promise with meta for new action or false if action with same meta.id was already in store.

MemoryStore#addReason(reasons, criteria)

Add reasons to metadata of actions, which are already in the store.

Reasons, which action already has, should not be duplicated.

Returns Promise. Promise when adding will be finished.

MemoryStore#byId(id)

Return action by action ID.

Returns Promise<[null, null] | [Action, ServerMeta]>. Promise with array of action and metadata.

MemoryStore#changeMeta(id, diff)

Change action metadata.

Returns Promise<boolean>. Promise with true if metadata was changed or false on unknown ID.

MemoryStore#clean()

Remove all data from the store.

Returns Promise. Promise when cleaning will be finished.

MemoryStore#get(opts?)

Return a Promise with first page. Page object has entries property with part of actions and next property with function to load next page. If it was a last page, next property should be empty.

This tricky API is used, because log could be very big. So we need pagination to keep them in memory.

Returns Promise<LogPage>. Promise with first page.

MemoryStore#getLastAdded()

Return biggest added number in store. All actions in this log have less or same added time.

Returns Promise<number>. Promise with biggest added number.

MemoryStore#getLastSynced()

Get added values for latest synchronized received/sent events.

Returns Promise<LastSynced>. Promise with added values

MemoryStore#remove(id)

Remove action from store.

Returns Promise<false | [Action, ServerMeta]>. Promise with entry if action was in store.

MemoryStore#removeReason(reasons, criteria, callback)

Remove reasons from action’s metadata and remove actions without reasons.

Returns Promise. Promise when cleaning will be finished.

MemoryStore#setLastSynced(values)

Set added value for latest synchronized received or/and sent events.

Returns Promise. Promise when values will be saved to store.

ServerConnection#connected

Is connection is enabled.

Type: boolean.

ServerConnection#destroy

Disconnect and unbind all even listeners.

Type: () => void.

ServerConnection#textMode

Whether to use text JSON protocol instead of binary. Always true for WsConnection, can change in WsBinaryConnection.

Type: boolean.

ServerConnection#ws

WebSocket connection instance

Type: WebSocket.

ServerConnection#connect()

Start connection. Connection should be in disconnected state from the beginning and start connection only on this method call.

This method could be called again if connection moved to disconnected state.

Returns Promise. Promise until connection will be established.

ServerConnection#disconnect(reason?)

Finish current connection.

ServerConnection#on(event, listener)

Returns Unsubscribe.

ServerConnection#send(message)

Send message to connection.

AbstractActionCreator#type

Type: string.

ActionCreator#match

Type: (action: Action) => action is CreatedAction.

ActionCreator#type

Type: string.

BaseServer#clientIds

Connected client by client ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient>.

BaseServer#connected

Connected clients.

for (let client of server.connected.values()) {
  console.log(client.remoteAddress)
}

Type: Map<string,ServerClient>.

BaseServer#env

Production or development mode.

if (server.env === 'development') {
  logDebugData()
}

Type: "development" | "production".

BaseServer#log

Server actions log.

server.log.each(finder)

Type: ServerLog.

BaseServer#logger

Console for custom log records. It uses pino API.

server.on('connected', client => {
  server.logger.info(
    { domain: client.httpHeaders.domain },
    'Client domain'
  )
})

Type: { debug: LogFn, error: LogFn, fatal: LogFn, info: LogFn, warn: LogFn }.

BaseServer#nodeId

Server unique ID.

console.log('Error was raised on ' + server.nodeId)

Type: string.

BaseServer#nodeIds

Connected client by node ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient>.

BaseServer#options

Server options.

console.log('Server options', server.options.subprotocol)

Type: BaseServerOptions.

BaseServer#subscribers

Clients subscribed to some channel.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: { }.

BaseServer#userIds

Connected client by user ID.

Do not rely on this data, when you have multiple Logux servers. Each server will have a different list.

Type: Map<string,ServerClient[]>.

BaseServer#addClient(connection)

Add new client for server. You should call this method manually mostly for test purposes.

server.addClient(test.right)

Returns number. Client ID.

BaseServer#auth(authenticator)

Set authenticate function. It will receive client credentials and node ID. It should return a Promise with true or false.

server.auth(async ({ userId, cookie }) => {
  const user = await findUserByToken(cookie.token)
  return !!user && userId === user.id
})

BaseServer#channel(pattern, callbacks, options?)

Define the channel.

server.channel('user/:id', {
  access (ctx, action, meta) {
    return ctx.params.id === ctx.userId
  }
  filter (ctx, action, meta) {
    return (otherCtx, otherAction, otherMeta) => {
      return !action.hidden
    }
  }
  async load (ctx, action, meta) {
    const user = await db.loadUser(ctx.params.id)
    ctx.sendBack({ type: 'USER_NAME', name: user.name })
  }
})

BaseServer#debugError(error)

Send runtime error stacktrace to all clients.

process.on('uncaughtException', e => {
  server.debugError(e)
})

BaseServer#destroy()

Stop server and unbind all listeners.

afterEach(() => {
  testServer.destroy()
})

Returns Promise. Promise when all listeners will be removed.

BaseServer#drain(clientId)

Wait until the client will confirm all actions, which were sent to it.

Use it to send a long history without loading it all into the memory: the client’s speed will limit how fast you read the database.

while (await server.drain(clientId)) {
  let page = await cursor.next(100)
  if (!page.length) break
  server.log.add(page.map(i => [i.action, { clients: [clientId] }]))
}

Returns Promise<boolean>. Promise with false if the client was disconnected.

BaseServer#handleClient(ws, req)

Handle WebSocket connection explicitly

This is a low-level method allowing to integrate Logux server with an existing server

fastify.get('/', { websocket: true }, (socket, req) => {
  loguxServer.handleClient(socket, req)
})

BaseServer#http(method, url, listener)

Add non-WebSocket HTTP request processor.

server.http('GET', '/auth', (req, res) => {
  let token = signIn(req)
  if (token) {
    res.setHeader('Set-Cookie', `token=${token}; Secure; HttpOnly`)
    res.end()
  } else {
    res.statusCode = 400
    res.end('Wrong user or password')
  }
})

BaseServer#listen()

Start WebSocket server and listen for clients.

Returns Promise. When the server has been bound.

BaseServer#on(event, listener)

Returns Unsubscribe.

BaseServer#otherChannel(callbacks)

Set callbacks for unknown channel subscription.

server.otherChannel({
  async access (ctx, action, meta) {
    const res = await phpBackend.checkChannel(ctx.params[0], ctx.userId)
    if (res.code === 404) {
      this.wrongChannel(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
})

BaseServer#otherType(callbacks)

Define callbacks for actions, which type was not defined by any Server#type. Useful for proxy or some hacks.

Without this settings, server will call Server#unknownType on unknown type.

server.otherType(
  async access (ctx, action, meta) {
    const response = await phpBackend.checkByHTTP(action, meta)
    if (response.code === 404) {
      this.unknownType(action, meta)
      return false
    } else {
      return response.body === 'granted'
    }
  }
  async process (ctx, action, meta) {
    return await phpBackend.sendHTTP(action, meta)
  }
})

BaseServer#process(action, meta?)

Add new action to the server and return the Promise until it will be resend to clients and processed.

Returns Promise<ServerMeta>. Promise until new action will be resend to clients and processed.

BaseServer#sendAction(action, meta)

Send action, received by other server, to all clients of current server. This method is for multi-server configuration only.

server.on('add', (action, meta) => {
  if (meta.server === server.nodeId) {
    sendToOtherServers(action, meta)
  }
})
onReceivingFromOtherServer((action, meta) => {
  server.sendAction(action, meta)
})

Returns void | Promise.

BaseServer#sendOnConnect(loader)

Change a way how server loads actions history for the client.

server.sendOnConnect(async (ctx, lastSynced) => {
  return db.loadActions({ user: ctx.userId, after: lastSynced })
})

Actions should be returned from the newest one to the oldest one. The list will be split into messages by the syncBatch option, but it is loaded into the memory as a whole. For a big history send it by pages with Context#sendBack() and Context#drain() instead.

Set meta.added to let the client ask only for newer actions after the reconnect: the biggest added will be sent as the sync position.

BaseServer#subscribe(nodeId, channel)

Send logux/subscribed if client was not already subscribed.

server.subscribe(ctx.nodeId, `users/${loaded}`)

BaseServer#type(actionCreator, callbacks, options?)