client doesn't receives if *args is in the event function parameters

[akshay7892]

Hello Miguel, first of all , a ton of thanks to you for creating such amazing product and making it awailable for the community, I really appreciate your efforts. I'm a newbie in python world so please bear with me. I'm trying to provide as much info as I can, to help you to help me solve this issue.
I'm getting live data from a broker's websocket, wrapped inside a python class which is stored in another file named "ws" and is imported into this file. The Broker has provided two functions socket and custom_message code for them is as follows.

socket

def socket(access_token):
    data_type = "symbolData"
    symbol =["NSE:SBIN-EQ"]
    fs = ws.FyersSocket(access_token=access_token,run_background=False,log_path="/home/Downloads/")
    fs.websocket_data = custom_message
    fs.subscribe(symbol=symbol,data_type=data_type)
    fs.keep_running()

custom_message

def custom_message(msg):
    print (f"Custom:{msg}")

As you can see the data from brokers data socket is getting set into custom_message function but I want to stream this data to a charting library so needed a websocket which could work with above python functions and thanks to you I had python socketio. So I created an ASGI websocket watching your videos.
The msg in received by custom_message is a python list containing a dictionary. As both the functions socket and custom_message are synchronous and the ASGI requires asyn functions, I needed a bridge that could connect these sync function to event async function. So I made custom_message function recive the message and forward it to async function. I implemented it as follows

import socketio, asyncio, config
from fyers_api.Websocket import ws
from Access_Token import access_token

sio = socketio.AsyncServer(async_mode='asgi',logger=True, engineio_logger=True,cors_allowed_origins=('*'))

app=socketio.ASGIApp(sio,static_files={
	'/':'./public/'
})

access_token = config.client_id+":"+access_token

@sio.event
async def connect(sid, environ):
	print(sid,"connected")

@sio.event
async def disconnect(sid):
	print(sid,"[disconnected]")

@sio.event
async def sum(sid,data):
	result = data['numbers'][0] + data['numbers'][1]
	await sio.emit('sum_result',{'result':result}, to=sid)
	await sio.emit('reconnect',{'result':1}, to=sid)
	print(sid,result)


def socket(sid,fysymbol):
	symbol =["NSE:SBIN-EQ"]
	data_type = "symbolData"
	fs = ws.FyersSocket(sid,fysymbol,access_token=access_token,run_background=False,log_path="/home/akshay/Documents/pytrade/log/")
	fs.websocket_data = custom_message
	fs.subscribe(symbol=symbol,data_type=data_type)
	fs.keep_running()

def custom_message(sid,data,*args):
	asyncio.run(SubAdd(sid,data,*args))
	
@sio.event
async def SubAdd(sid,data,*args): ########This is what is causing issue
	await sio.emit('reconnect',{'result':1}, to=sid)
	if args:
		for arg in args:
			emit_data = str(str(0)+"~"+str(arg['symbol'].replace("-","~").replace(":","~"))+"~"+str(arg['timestamp'])+"~"+str(arg['min_open_price'])+"~"+str(arg['min_high_price'])+"~"+str(arg['min_low_price'])+"~"+str(arg['min_close_price'])+"~"+str(arg['min_volume']))
			print(f"emit_data={emit_data}")
			await sio.emit("m",emit_data, to=sid)
	sid = sid
	fysymbol = data['fysymbol']
	print("SubAdd isConnected")
	#socket(sid,fysymbol)
	
@sio.event
async def SubRemove(sid,data):
	fysymbol,resolution = data['fysymbol'],data['resolution']

	fyersSocket.unsubscribe(symbol=symbol)
	symbol.remove(fysymbol)
	await sio.emit('UnSubscribed',data['fysymbol'], to=sid)
	print(sid,"[UnSubscribed]",data['fysymbol'])

Which is working fine if we look at the server's logs in the terminal e.g.

received event "sum" from GOGFG4v_gCrobHcFAAAB [/]
emitting event "sum_result" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["sum_result",{"result":3}]```

and

 ```emitting event "m" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652259912~471.1~471.15~470.85~471.15~9400"]

Terminal Logs

uvicorn app:app --port 9999 --reload
INFO:     Will watch for changes in these directories: ['/home/akshay/Documents/pytrade/pytrade']
INFO:     Uvicorn running on http://127.0.0.1:9999 (Press CTRL+C to quit)
INFO:     Started reloader process [12498] using watchgod
Server initialized for asgi.
INFO:     Started server process [12509]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
4dacEB52_RzDB83fAAAA: Sending packet OPEN data {'sid': '4dacEB52_RzDB83fAAAA', 'upgrades': [], 'pingTimeout': 20000, 'pingInterval': 25000}
4dacEB52_RzDB83fAAAA: Received request to upgrade to websocket
INFO:     ('127.0.0.1', 40320) - "WebSocket /socket.io/" [accepted]
4dacEB52_RzDB83fAAAA: Upgrade to websocket successful
INFO:     connection open
4dacEB52_RzDB83fAAAA: Received packet MESSAGE data 0
GOGFG4v_gCrobHcFAAAB connected
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 0{"sid":"GOGFG4v_gCrobHcFAAAB"}
4dacEB52_RzDB83fAAAA: Received packet MESSAGE data 2["sum",{"numbers":[1,2]}]
received event "sum" from GOGFG4v_gCrobHcFAAAB [/]
emitting event "sum_result" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["sum_result",{"result":3}]
emitting event "reconnect" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["reconnect",{"result":1}]
GOGFG4v_gCrobHcFAAAB 3
4dacEB52_RzDB83fAAAA: Received packet MESSAGE data 2["SubAdd",{"fysymbol":"NSE:SBIN-EQ","resolution":"1"}]
received event "SubAdd" from GOGFG4v_gCrobHcFAAAB [/]
emitting event "reconnect" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["reconnect",{"result":1}]
SubAdd isConnected
sid,fysymbol= ('GOGFG4v_gCrobHcFAAAB', 'NSE:SBIN-EQ')
emitting event "reconnect" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["reconnect",{"result":1}]
emit_data=0~NSE~SBIN~EQ~1652259910~471.1~471.15~470.85~471.1~6627
emitting event "m" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652259910~471.1~471.15~470.85~471.1~6627"]
emitting event "reconnect" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["reconnect",{"result":1}]
emit_data=0~NSE~SBIN~EQ~1652259911~471.1~471.15~470.85~471.15~6757
emitting event "m" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652259911~471.1~471.15~470.85~471.15~6757"]
emitting event "reconnect" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["reconnect",{"result":1}]
emit_data=0~NSE~SBIN~EQ~1652259912~471.1~471.15~470.85~471.15~9400
emitting event "m" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652259912~471.1~471.15~470.85~471.15~9400"]
emitting event "reconnect" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["reconnect",{"result":1}]
emit_data=0~NSE~SBIN~EQ~1652259913~471.1~471.15~470.85~471.15~9450
emitting event "m" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652259913~471.1~471.15~470.85~471.15~9450"]
emitting event "reconnect" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["reconnect",{"result":1}]
emit_data=0~NSE~SBIN~EQ~1652259914~471.1~471.15~470.85~471.15~9520
emitting event "m" to GOGFG4v_gCrobHcFAAAB [/]
4dacEB52_RzDB83fAAAA: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652259914~471.1~471.15~470.85~471.15~9520"]

The client receives the sum_result and reconnect message from sum event. However, it isn't receiving the (second) reconnect and m message in the SubAdd event which is evident from message logs in devtools.

Logs of devtools

So what and where should I make the changes so that client whould receive all messages from all events.
Thank You!!

client.js

const socket = io('http://localhost:9999', {
     transports: ['websocket', 'polling', 'flashsocket']
});


const channelToSubscription = new Map();
console.log('ChannelToSubscription=',channelToSubscription ),

socket.on('connect', () => {
	console.log('[socket] Connected');
	socket.emit('sum',{numbers:[1,2]});
});

socket.on('sum_result', (data) =>{
    console.log(data.result);

});

socket.on('reconnect', (data) =>{
    console.log(data.result);

});

socket.on('disconnect', (reason) => {
	console.log('[socket] Disconnected:', reason);
});

socket.on('error', (error) => {
	console.log('[socket] Error:', error);
});


socket.on('m', data => {
	console.log('[socket] Message:', data);
	const [
		eventTypeStr,
		exchange,
		fromSymbol,
		toSymbol,
		tradeTimeStr,
		openPrice,
		highPrice,
		lowPrice,
		closePrice,
		volumeStr,
	] = data.split('~');
	//console.log(lowPrice);

	//if (parseInt(eventTypeStr) !== 0) {
	//	// skip all non-TRADE events
	//	return;
	//}
	const tradePrice = parseFloat(openPrice);
	const tradeTime = parseInt(tradeTimeStr)*1000;
	const openStr = parseFloat(openPrice);
	const highStr = parseFloat(highPrice);
	const lowStr = parseFloat(lowPrice);
	const closeStr = parseFloat(closePrice);
	const volume_Str = parseFloat(volumeStr);

	//console.log(volume_Str);

	const channelString = `${exchange}:${fromSymbol}-${toSymbol}`;
	const subscriptionItem = channelToSubscription.get(channelString);
	if (subscriptionItem === undefined) {
		return;
	}

[miguelgrinberg]

I'm sorry, but this is very difficult for me to understand, so I cannot give you specific details. What I can tell you is that you cannot mix sync and async code in the same application. If your input WebSocket uses sync code, then you should use a sync SocketIO server, as a mixture will cause blocking on the async side.

Also, please do not file issues on this project when you need help. The issues board is for bugs. Use the discussions board when you have questions. Thanks.

[akshay7892]

Hello Miguel, apologies for posting in the issues borad. I tried using sync SocketIO server before using async one, but it also has the same issue. Server emits the data but the client never receives it. I cant provide the server logs today as websocket from broker closes at 3.30 Indian standerd time and now its 4.40 IST . Will provide the logs tomorrow.

[akshay7892]

Hello Miguel,
The issue has been resolved and it is working fine now in sync mode. I found the solution in one of the discussion topic and to try it out added eventlet monkeypatch as you mentioned in #899. Thanks for your help and patience.

1waWVU6PV0g50at_AAAB 3
1waWVU6PV0g50at_AAAB disconnected
127.0.0.1 - - [12/May/2022 13:24:08] "GET /socket.io/?EIO=4&transport=websocket HTTP/1.1" 200 0 2.632063
(8978) accepted ('127.0.0.1', 52518)
Cyko-gF9ZIxGu3CRAAAC: Sending packet OPEN data {'sid': 'Cyko-gF9ZIxGu3CRAAAC', 'upgrades': [], 'pingTimeout': 20000, 'pingInterval': 25000}
Cyko-gF9ZIxGu3CRAAAC: Received request to upgrade to websocket
Cyko-gF9ZIxGu3CRAAAC: Upgrade to websocket successful
Cyko-gF9ZIxGu3CRAAAC: Received packet MESSAGE data 0
2ezBoDNN_alLN6tcAAAD connected
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 0{"sid":"2ezBoDNN_alLN6tcAAAD"}
Cyko-gF9ZIxGu3CRAAAC: Received packet MESSAGE data 2["sum",{"numbers":[1,2]}]
received event "sum" from 2ezBoDNN_alLN6tcAAAD [/]
emitting event "sum_result" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["sum_result",{"result":3}]
emitting event "message" to all [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["message",{"result":3}]
2ezBoDNN_alLN6tcAAAD 3
Cyko-gF9ZIxGu3CRAAAC: Received packet MESSAGE data 2["SubAdd",{"fysymbol":"NSE:SBIN-EQ","resolution":"1"}]
received event "SubAdd" from 2ezBoDNN_alLN6tcAAAD [/]
sid,fysymbol in ws.py= ('2ezBoDNN_alLN6tcAAAD', 'NSE:SBIN-EQ')
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342055~462.3~462.5~462.15~462.45~17315"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342058~462.3~462.5~462.15~462.45~17998"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342062~462.3~462.5~462.15~462.35~20181"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342064~462.3~462.5~462.15~462.2~21300"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342067~462.3~462.5~462.05~462.25~22088"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342068~462.3~462.5~462.05~462.05~22243"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342069~462.3~462.5~462.05~462.2~22256"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342074~462.3~462.5~462.0~462.15~22961"]
Cyko-gF9ZIxGu3CRAAAC: Sending packet PING data None
Cyko-gF9ZIxGu3CRAAAC: Received packet PONG data 
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342078~462.3~462.5~462.0~462.0~24775"]
emitting event "m" to 2ezBoDNN_alLN6tcAAAD [/]
Cyko-gF9ZIxGu3CRAAAC: Sending packet MESSAGE data 2["m","0~NSE~SBIN~EQ~1652342082~462.3~462.5~461.95~462.0~26681"]