Broadcasting message from grpc server to all/some connected clients in python

Question

i am learning how to use grpc streams to exchange messages between clients and server in python. I found a base example that enables the simple message sending between server and client. I am trying to modify it so that i could keep track of all the clients connected to the grpc server (on the server side) and could do two things: 1) broadcast from server to all clients, 2) send message to a particular connected client.

Here is the .proto file

syntax = 'proto3';

service Scenario {
    rpc Chat(stream DPong) returns (stream DPong) {}
}

message DPong {
    string name = 1;
}

And here is the client.py that creates a daemon process to listen for incoming messages and waits for stdin for any outgoing messages

import threading

import grpc
import time
import scenario_pb2_grpc, scenario_pb2

# new changes
msgQueue = queue.Queue()
def run():
    channel = grpc.insecure_channel('localhost:50052')
    stub = scenario_pb2_grpc.ScenarioStub(channel)
    print('client connected')
    global queue

    def inputStream():
        while 1:
            msg = input('>>Enter message\n>>')
            yield scenario_pb2.DPong(name=msg)

    input_stream = stub.Chat(inputStream())
    def read_incoming():
        while 1:
            print('receivedFromServer: {}\n>>'.format(next(input_stream).name))

    thread = threading.Thread(target=read_incoming)
    thread.daemon = True
    thread.start()
    while 1:
        time.sleep(1)

if __name__ == '__main__':
    print('client starting ...')
    run()

Below is the server.py

import random
import string
import threading
import grpc
import scenario_pb2_grpc
import scenario_pb2
import time
from concurrent import futures

clientList = []
class Scenario(scenario_pb2_grpc.ScenarioServicer):

    def Chat(self, request_iterator, context):
        clients = []
        def stream():
            while 1:
                time.sleep(1)
                msg = input('>>Enter message\n>>')
                for i in clientList:
                  yield msg

        output_stream = stream()
        def read_incoming():
            while 1:
                received = next(request_iterator).name
                if (context,request_iterator) not in clientList:
                  clientList.append((context, request_iterator))
                print('receivedFromClient: {}'.format(received), len(clientList))

        thread = threading.Thread(target=read_incoming)
        thread.daemon = True
        thread.start()
        while 1:
            msg = output_stream
            yield scenario_pb2.DPong(name=next(msg))

if __name__ == '__main__':
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    scenario_pb2_grpc.add_ScenarioServicer_to_server(
        Scenario(), server)
    server.add_insecure_port('[::]:50052')
    server.start()
    print('listening ...')
    while 1:
        time.sleep(1)

So far, i have tried to maintain a list object clientList that contains the context & request_iterator object of the client, and is updated every time a new client joins the server. But how do i set these object from the clientList before sending out an outgoing message? I have tried to iterate the list but the server sends the message to the same client (the last client heard from) a number of times instead of sending it to all the clients once.

Any help is highly appreciated!

Answer 1

You can spin up multiple ports in one application.

gRPC can be running in port 50011 and flask with socket.io can be running in port 8080

with python, you can use the flask framework and flask_socketio library in your server.py

eg server.py

from flask import Flask
from flask_socketio import SocketIO, emit

app = Flask(__name__)
socketio = SocketIO(app)

@app.route('/')
def index():
    return "Hello, World!"

if __name__ == '__main__':
    app.run(port=8080)
    app.run(debug=True)
    socketio.run(app)

instead of using gRPC streaming API, use WebSocket to broadcast to all connected clients and specific/selected clients using rooms.

eg

@socketio.on('message')
def handle_message(data):
    // logic to send large data in chunks the logic should call the 
    // emit function in socket.io and emit an event that send the large
    // data in chunks eg emit('my response', chunkData)

gRPC is primarily built for one client request and response and WebSocket is for multiple clients.

Answer 2

This is certainly possible. The problem that you're running into here is that each call to Scenario.Chat on the server side corresponds to a single client connection. That is, this function is called when the streaming RPC starts and as soon as the function exits, the RPC ends.

So if you want n connected clients, you'll need n instances of Scenario.Chat running concurrently, each on its own thread. This does mean that the number of concurrently connected clients is limited by the size of the threadpool with which you instantiate your server.

So, let's say you have n threads in your server process dedicated to maintaining client connections. Then you need another n+1th thread (perhaps the main thread) determining when the server will broadcast a message to all clients (maybe by looking for input from STDIN?). When this extra thread determines that a message should be broadcast, it needs to communicate this intent to all of the threads maintaining connections to a client. There are many ways to make this happen. A threading.Condition and a global collections.deque, or a collections.deque per client connection (somewhat like channels between goroutines) would be two ways. The tricky bit here is ensuring that each client connection will receive the message regardless of how long the client connection thread takes to wake up and how many messages the n+1th thread decides to send in the interim.

If this is still unclear, I can follow up with some actual code demonstrating the idea.