socket.disconnect(code, msg);. We have certain timeout for authentication, so user can't stay connected unauthenticated and rule that was checking this condition was broken due to changed flow, but its still a mystery to me why it throws such error if this .disconnect() occurs way after authentication has passed and JWT received by the client.
Hello, I am new to SC, which seems pretty promising.
I am trying to allow or deny connection to the server depending on data sent by the client. I am using a handshake middleware for this which looks at the query parameters sent when creating the client. This seems to work fine, but I can't manage to receive the error message on the client side that was generated using action.block(err) in the server side middleware. All I get on the client side is an error of type "SocketProtocolError" with the generic message "Socket hung up".
Here's the relevant code inside the handshake middleware:
for await (let action of middlewareStream) {
const err = new Error('No ID provided');
err.name = 'MissingParameter';
action.block(err);
continue;
}And here's the listener on the client side:
for await (let data of this.socket.listener('error')) {
console.log('error', data);
};I also tried listening to other events such as connectAbort, close and disconnect, but those are not called.
Am I doing something wrong? Should I not be doing this during the handshake?
Any help would be highly appreciated.
Thank you in advance.
MissingParameter response.
this.socket.listener('error')
(async () => {
for await (let { error } of socket.listener('error')) {
console.error(error);
}
})();HANDSHAKE_WS middleware is special because it happens before the underlying WebSocket has been created (at the HTTP/WS handshake stage). If you block the connection by passing an error to the next(err) callback, the error string will show up in your browser's developer panel (Network tab in Chrome) but there is no way to handle this error in your code. If you try to listen to the 'connectAbort' event, the error code will always be 1006 and you won't be able to get any additional information about it. This is a limitation of the WebSocket RFC itself, For this reason, blocking connections with HANDSHAKE_WS mostly makes sense for quickly and efficiently shutting down malicious connections. If you want a more client-friendly way to kill a connection, you should use the HANDSHAKE_SC middleware instead.
socket.exchange.subscribe(channelName), the second socket exchange is subscribe to the same channel as well (return true when call socket.exchange.isSubscribed(channelName)). And when i perform unsubscribe on one socket, the other one unsubscribe too. I acknowledge that this behaviour only happens in version 15+.Does anyone have any idea that I can do a subscribe channel on the server side (not the client) that is separate between each socket connection?
@SloppyShovel here is my current code
export const subscribeChannel = async (socket, channelName) => {
const remote = `${socket.remoteAddress}:${socket.remotePort}`
let channel
if (socket.exchange.isSubscribed(channelName)) {
channel = socket.exchange.channel(channelName)
} else {
channel = socket.exchange.subscribe(channelName)
}
for await (const data of channel) {
const userId = socket.authToken?.uid || ANONYMOUS_USER_ID
if (socket.state !== socket.CLOSED) {
console.log(`User ${userId} watch channel ${channelName}'s ${JSON.stringify(data)} on ${remote}`)
}
if (socket.state !== socket.CLOSED) socket.transmit('notification', data)
}
}
export const unsubscribeChannel = async (socket, channelName, msg) => {
const userId = socket.authToken?.uid || ANONYMOUS_USER_ID
const publishData = { uid: userId, msg }
const subscriberClients = socket.exchange._broker._clientSubscribers[channelName] || {}
Object.values(subscriberClients).forEach((client) => {
client.transmit('#publish', {
event: 'notification',
channel: channelName,
data: publishData,
})
})
const channel = socket.exchange.channel(channelName)
channel.closeAllListeners()
socket.exchange.unsubscribe(channelName)
}I want all user subcribe to the same channel, so that when a new user joins or exits other users would be notified. I want to do this separately from the client side since client-side channels are unmanaged on the server side.
(async () => {
for await (let { socket, channel, channelOptions } of agServer.listener('subscription')) {
// client subscribed to a channel
}
})();(async () => {
for await (let { socket, channel} of agServer.listener('unsubscription')) {
// client unsubscribed from a channel
}
})();
I am currently building a solution with SC for distributing data into channels were the data is sent by SC client.
So basically, when the server is getting a data on some specific channels the server to loop on the data and distribute it into unique channel.
Now, SC is scalable means I can have multiple workers running at the same time, which each one of them will subscribe to the "supplying" channels data but the distributing needs to happened only once.
My question is if someone ever build similar use case, what is the best / elegant option of making sure that the data will be distribute only once.
My current idea is using unique locks with TTL using couchbase,
Means all SC workers will still subscribe to the channels, each one will try to distribute but only one will succeed due to lock.
Here is some real life example
Imagine my supplier SC Client sent the data in that structure into "stocks" channel:
{
"event": "update",
"id": 123456,
"data": [
{
"id": 1,
"name": "apple",
"type": "fruit",
"stock": 10
},
{
"id": 2,
"name": "chaires",
"type": "furniture",
"stock": 156
}
]
}My server side subscribe to 'stocks' channel and needs to loop the "data" and sent every object into different channels what I am calling "distribution"
distributing channels will be unique by the object "id","name" and "type"
So the it will like that:
agServer.exchange.transmitPublish(`stocks:${id}`, object);
agServer.exchange.transmitPublish(`stocks:${name}`, object);
agServer.exchange.transmitPublish(`stocks:${type}`, object);The thing is that only 1 worker can take care of it otherwise there will be multiple distributions into those channels.
Brokers are the bridge between Workers
State is like etcd but simple, like the name states, it holds the state of the cluster, so the number of workers, brokers and there addresses.
the answer is in another question or statement "The thing is that only 1 worker can take care of it otherwise there will be multiple distributions into those channels."
if your fighting so much with the architecture of the framework your doing it 99.9% of the time wrong.
Yes, my use case is a bit different,
I have 1 client working as my "supplier" -> he's pumping real time events from external system and publishing to my SC server in specific channels like "stocks" as my example above.
So now I will just build my distributors as clients and it will allow me to make sure to have 1 source of distribution and also scalability because I can start many clients but each one will take care of distrusting different channels.
I agree, the whole system won't be scalable because I have one "supplier" the can break the whole thing, but it's doesn't critical in my case, even if it fails for certain time it's acceptable and won't be critical for the operation of my business logic.
"you can't integrate the "supplier" into the server?"
Unfortunately not, my supplier is a sort of a plugin injected into the external system, also in C++, from the external system point of view it won't be a good idea to distribute the data from the supplier client due the possibility of affecting the external system performance (this system is scalable).
I already built the client distributors seems to works as expected,
Thanks a lot for your help if you have any other suggestion I would love to hear :)