AWS API Gateway Websocket Asynchronous Notifications Pusher

Overview

AWS API Gateway Websocket Asynchronous Pusher

Illustration

Fast AWS API Gateway websockets notifications' pusher using Python AsyncIO for managing asynchronous and concurrent non-blocking IO calls to DynamoDB connections store and API Gateway management API. making it suitable for broadcasting messages to multiple users with a fast and cost-effective approach.

You can test the Pusher with this demo application or use the live demo at https://sumu.kodhive.com

Features

  • Multiple sources: capable of receiving notification requests from SQS and SNS, which make it suitable for processing single notification request from SNS and high number of notification requests in batches when polling from SQS

  • Multicast Notifications: the pusher can send messages to only a subset of users (eg: users in a chat room).

  • Broadcast Notifications: sometimes you just want to send the same message for all connected users (eg: broad announcements)

  • Exclusion Notifications: the pusher can broadcast messages to all users except a list of excluded users (eg: online/offline presence events can be sent to all except the originator)

  • Stale Connections Pruning: capable of detecting and deleting stale connections from DynamoDB connections store in case API Gateway missed cleaning them.

  • Asynchronous Processing: the pusher is using AsyncIO to notify multiple users/connections concurrently to not wait for inflight requests to DynamoDB/API Gateway so you don't pay AWS the waiting time 😎

  • Batch Processing: when using SQS as event source the Pushed will be able to process batches of notification requests, also concurrently.

  • Duplicate Users Detection: able to detect duplicate users in a notification requests and make them unique set of users. to avoid double notifications.

Requirements

Illustration

  • DynamoDB Connections Store: a DynamoDB table where users connections are stored, It should have a hash key user_id and a range key connection_id, where every user can have multiple connections and the Pusher can retrieve specific user connections by querying with the hash key.

  • AWS API Gateway Websockets API: an already configured Websocket API Gateway. the Pusher will use the connection management endpoint passed as environment variable to post messages to users.

  • SNS Topic: the SNS Topic that the pusher will subscribe to for notifications requests.

  • SQS Queue: the SQS Queue that the pusher will poll notifications requests from.

Notification Requests

Backend applications can send notifications to AWS API Gateway Websocket connected users by sending a notification request to the service integrated with the Pusher (SNS|SQS), notifications requests should meet the following format:

For Multicast notifications, the message should be a JSON String that contains the list of users and the actual data:

import json

message = {
    "users": ["783304b1-2320-44db-8f58-09c3035a686b", "a280aa41-d99b-4e1c-b126-6f39720633cc"],
    "data": {"type": "notification", "message": "A message sent to multiple user"}
}
message_to_send = json.dumps(message)

For broadcast notifications, the same but do not provide users list or provide an empty users list:

import json

message = {
    "data": {"type": "announcement", "message": "A broadcast to all users"}
}
message_to_send = json.dumps(message)

For exclusion notifications, instead of providing users list, provide a list of excluded users:

import json

message = {
    "exclude_users": ["783304b1-2320-44db-8f58-09c3035a686b"],
    "data": {
      "type": "announcement", 
      "message": {
        "user_id": "783304b1-2320-44db-8f58-09c3035a686b", 
        "status": "OFFLINE"
      }
    }
}
message_to_send = json.dumps(message)

Integration

The Pusher can accept notifications requests from multiple sources like SQS and SNS. every source has its advantages and disadvantages when it comes to speed and cost.

Notification requests through SNS

The Pusher can subscribe to notifications SNS Topic, and whenever a backend applications Publish notification requests to SNS, the later will quickly notify the Pusher by sending the notification request to the subscribed Pusher Lambda.

This will result in a fast delivery because this approach does not introduce a polling mechanism and SNS will notify the Pusher whenever a notification request is available. however, at scale SNS will trigger a Pusher Lambda Function for every notification request and given that the Lambda Function Concurrent Invocations Limit is 1,000 per account (Can be increased to 100,000 by support-ticket) notification requests will be throttled for large applications.

Publish to SNS when you have small application with few users

import os
import json
import time
import boto3

message = {
    "users": ["783304b1-2320-44db-8f58-09c3035a686b", "a280aa41-d99b-4e1c-b126-6f39720633cc"],
    "data": {
        "type": "notification",
        "message": {
            "text": "Your order has been fulfilled!",
            "timestamp": int(time.time())
        }
    }
}
boto3.client("sns").sns.publish(
    TargetArn=os.environ["NOTIFICATIONS_TOPIC_ARN"],
    Message=json.dumps(message),
)

Sending notification requests through SQS

Unlike SNS, when sending notifications to SQS queue, the Pusher Lambda Function event source can be configured to poll notification requests from the SQS Queue, and it will periodically poll notification requests from the Queue using Polling Technique.

This will result in notifications requests to be processed in batches, which comes with many benefits:

  • Fewer lambda function executions, to not reach the Lambda Concurrent Execution Limit.
  • As the pusher uses AsyncIO, it will be able to process batches of SQS Records concurrently.
  • Low cost thanks to SQS batches and fewer Lambda Executions.

You can meet the same speed and performance of SNS if you set the SQS queue receive_wait_time_seconds to 0. this will make the Lambda Function do Short Polling instead of Long Polling. and it will receive the notifications requests immediately after being visible on SQS queue.

Send to SQS when you have a large application with millions of users

import os
import json
import time
import boto3

message = {
    "users": ["783304b1-2320-44db-8f58-09c3035a686b", "a280aa41-d99b-4e1c-b126-6f39720633cc"],
    "data": {
        "type": "notification",
        "message": {
            "text": "Your order has been fulfilled!",
            "timestamp": int(time.time())
        }
    }
}
boto3.client("sqs").send_message(
  QueueUrl=os.environ.get("NOTIFICATIONS_QUEUE_URL"),
  MessageBody=json.dumps(message),
)

Deploy

Before deploying the pusher, version control it on your github account and then call the Terraform module shipped with the Pusher to provision the Pusher Lambda Function, and the Terraform module to create CI/CD pipeline:

module "pusher" {
  source      = "./infra"
  prefix      = local.prefix
  common_tags = local.common_tags

  agma_arn       = "${aws_apigatewayv2_api._.execution_arn}/${aws_apigatewayv2_stage._.name}/POST/@connections"
  apigw_endpoint = "https://live.kodhive.com/push"
  # Custom API Gateway Domain
  connections_table = {
    name = aws_dynamodb_table.connections.name
    arn  = aws_dynamodb_table.connections.arn
  }
  notifications_topic_arn = aws_sns_topic.notifications.arn
  notifications_queue_arn = aws_sqs_queue.notifications.arn
}

module "pusher_ci" {
  source      = "git::https://github.com/obytes/terraform-aws-lambda-ci.git//modules/ci"
  prefix      = "${local.prefix}-pusher-ci"
  common_tags = var.common_tags

  # Lambda
  lambda                   = module.pusher.lambda
  app_src_path             = "src"
  packages_descriptor_path = "src/requirements/lambda.txt"

  # Github
  pre_release  = true
  s3_artifacts = {
    arn    = aws_s3_bucket.artifacts.arn
    bucket = aws_s3_bucket.artifacts.bucket
  }
  github = {
    owner          = "obytes"
    webhook_secret = "not-secret"
    connection_arn = "arn:aws:codestar-connections:us-east-1:{ACCOUNT_ID}:connection/{CONNECTION_ID}"
  }
  github_repository = {
    name   = "apigw-websocket-pusher"
    branch = "main"
  }
  # Notifications
  ci_notifications_slack_channels = {
    info  = "ci-info"
    alert = "ci-alert"
  }
}
Owner
OBytes
We're a team of dreamers and makers who create products that your users will love
OBytes
SocketIO 转发台,保持 botoy 能力和插件功能的同时,透传其接口,以此使用更灵活、生态更好的技术进行开发

bot_sio_transfer SocketIO 转发台,保持 botoy 能力和插件功能的同时,透传其接口,以此使用更灵活、生态更好的技术进行开发 Usage 请参考 botoy 文档接入本插件 Example 考虑一种图文混排场景,如何从复杂结构内快速获取第一张图片的 url ? ( 假设图片

OPQ Open Source Community 3 Oct 18, 2021
Whatsapp Clone using django, django-channels and websocket

whatsapp-clone Whatsapp Clone using django, django-channels and websocket Features : Signup/Login One on One personal chat with other user Some screen

Anshu Pal 14 Dec 25, 2022
Websockify is a WebSocket to TCP proxy/bridge. This allows a browser to connect to any application/server/service. Implementations in Python, C, Node.js and Ruby.

websockify: WebSockets support for any application/server websockify was formerly named wsproxy and was part of the noVNC project. At the most basic l

noVNC 3.3k Jan 03, 2023
Tetri5 - Multiplayer Websocket Backend

Tetri5 - Multiplayer Websocket Backend This repository is the backend of the multiplayer portion of the Tetri5 game client. It uses the python websock

Giovani Rodriguez 1 Dec 10, 2022
WebSocket emulation - Python server

SockJS-tornado server SockJS-tornado is a Python server side counterpart of SockJS-client browser library running on top of Tornado framework. Simplif

Serge S. Koval 854 Nov 19, 2022
JINS MEME(2021年モデル)のJINS MEME LoggerをPythonのWebSocketサーバーで受信するサンプル

JINS-MEME-Python-WebSocketServer-Sample JINS MEME(2021年モデル)のJINS MEME LoggerをPythonのWebSocketサーバーで受信するサンプルです。 Logging Data 以下のデータに対応しています。 各データの定義はJIN

KazuhitoTakahashi 19 Apr 29, 2022
A Security Tool for Enumerating WebSockets

STEWS: Security Testing and Enumeration of WebSockets STEWS is a tool suite for security testing of WebSockets This research was first presented at OW

175 Jan 01, 2023
Async (trio) KuCoin minimal REST API + Websocket

Minimal Async KuCoin REST API + WebSocket using trio Coded by π ([email protected] TG: @p

Pi 2 Oct 23, 2022
This websocket program is for data transmission between server and client. Data transmission is for Federated Learning in Edge computing environment.

websocket-for-data-transmission This websocket program is for data transmission between server and client. Data transmission is for Federated Learning

9 Jul 19, 2022
Connects microservices through a mesh of websockets

WebMesh WebMesh is a WebSocket based communication library for microservices. It uses a WebSocket server based on wsproto that distributes clients on

Charles Smith 9 Apr 29, 2022
Burgeramt-appointments-websockets - Fetch Bürgeramt appointments and broadcast them via websockets

Bürgeramt appointment finder This server looks for Bürgeramt appointment every f

74 Dec 19, 2022
Websocket RPC and Pub/Sub for Python applications and microservices

wampy [whomp-ee] For a background as to what WAMP is, please see here. This is a Python implementation of WAMP using Gevent, but you can also configur

simon 121 Nov 22, 2022
Websocket 'broadcast' demo using FastAPI/Starlette

fastapi-websocket-broadcast An example of the familiar 'chat' websocket demo app, implemented in FastAPI / Starlette. Run with uvicorn app:app And th

Kit Thwaite 109 Nov 30, 2022
Library for easily creating and managing websockets.

Documentation coming in version 0.1.4 GitHub PyPI Discord Features Easy to use with object oriented syntax. Intellisense support with typehints and do

ZeroIntensity 0 Aug 27, 2022
Minecraft WebSocket

Minecraft-WebSocket Pythonでマインクラフトと通信します。 紹介動画 推奨設定 Minecraft Windows Edition (Education Edition) 1.17 以上 Python 3系(3.8.2で動作確認済み) 必要なモジュール ・asyncio ・w

Roii.py 2 Jul 07, 2022
Chat app for Django, powered by Django Channels, Websockets & Asyncio

Django Private Chat2 New and improved https://github.com/Bearle/django-private-chat Chat app for Django, powered by Django Channels, Websockets & Asyn

Bearle 205 Dec 30, 2022
A simple texts communication platform using pygame and sockets.

Python Tchat A simple texts communication platform using pygame and sockets. How to install ? You need a python 3.8.6 server with at least one opened

Yohann Boniface 7 Jan 23, 2022
wssh ("wish") is a command-line utility/shell for WebSocket inpsired by netcat.

wssh ("wish") is a command-line utility/shell for WebSocket inspired by netcat

Jeff Lindsay 256 Nov 16, 2022
Terminals served by tornado websockets

This is a Tornado websocket backend for the Xterm.js Javascript terminal emulator library. It evolved out of pyxterm, which was part of GraphTerm (as

Project Jupyter 332 Dec 27, 2022
An IPC based on Websockets, fast, stable, and reliable

winerp An IPC based on Websockets. Fast, Stable, and easy-to-use, for inter-communication between your processes or discord.py bots. Key Features Fast

Black Thunder 5 Aug 09, 2022