Serverless Websocket Setup

serverless-websocket.md

Serverless Websocket Setup

Install Client

npm install wscat -g

Connect

wscat -c <websocket url>

Serverless YAML Lambda Handler

ConnectHandler:
    handler: src/websocket/connect.handler
    events:
      - websocket:
          route: $connect

DisconnectHandler:
    handler: src/websocket/disconnect.handler
    events:
      - websocket:
          route: $disconnect

Node labmda handler for DynamoDB

Use APIGatewayProxyHandler (APIGatewayProxyEvent)

Add connection

connectionId = event.requestContext.connectionId
const timestamp = new Date().toISOString()

const item = {
  id: connectionId,
  timestamp
}

// Store in connection
await docClient.put({
  TableName: connectionsTable,
  Item: item
}).promise()

return {
  statusCode: 200,
  body: ''
}

Delete connection

connectionId = event.requestContext.connectionId
const key = {
  id: connectionId
}

// Store in connection
await docClient.delete({
  TableName: connectionsTable,
  Key: key
}).promise()

return {
  statusCode: 200,
  body: ''
}

Set ENV for Function Only

SendNotifications:
 environment:
  STAGE: ${self:provider.stage}
  API_ID:
   Ref: WebsocketsApi
 handler:
  src/lambda/sendNotifications.handler

Send Notifications

const connectionsTable = process.env.CONNECTIONS_TABLE
const stage = process.env.STAGE
const apiId = process.env.API_ID

const connectionParams = {
 apiVersion: "2018-11-29",
 endPoint: `${apiId}.execute-api.<change to region>.amazonaws.com/${stage}`
}

const apiGateway = new AWS.ApiGatewayManagementApi(connectionParams)

export const handler: S3Handler = async (event: S3Event) => {
  for (const record of event.Records) {
    const key = record.s3.object.key
    console.log('Processing S3 item with key: ', key)
  
    const connections = await docClient.scan({
      TableName: connectionsTable
    }).promise()
  
    const payload = {
      imageId: key
    }
    
    for (const connection of connections.Items) {
      await sendMessageToClient(connection.id, payload)
    }
  }
}

async function sendMessageToClient(connectionId, payload) {
  try {
    console.log('Sending message to a connection', connectionId)
    await apiGateway.postToConnection({
      ConnectionId: connectionId,
      Data: JSON.stringify(payload)
    }).promise()
  } catch (e) {
    console.log('Failed to send message', JSON.stringify(e))
    
    // Process stale connection
    if (e.statusCode = 410) {
      console.log('Stale connection')
      
      await docClient.delete({
        TableName: connectionsTable,
        Key: {
          id: connectionId
        }
      }).promise()
    }
  }
}