Skip to content

event-streaming-2 #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ttodua wants to merge 4 commits into
base: event-streaming
Choose a base branch
from
Draft

event-streaming-2 #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
738821d
rev
ttodua Mar 19, 2024
156048e
raw streams - compact
ttodua Mar 19, 2024
2ccb6df
transpilable
ttodua Mar 19, 2024
e7bd9f7
rename
ttodua Mar 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 29 additions & 1 deletion ts/src/base/Exchange.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1404,6 +1404,10 @@ export default class Exchange {
this.clients = this.clients || {};
if (!this.clients[url]) {
const onMessage = this.handleMessage.bind (this);
const onMessageWithStream = (client, message)=> {
this.streamProduce ('raw', message);
return onMessage (client, message);
};
const onError = this.onError.bind (this);
const onClose = this.onClose.bind (this);
const onConnected = this.onConnected.bind (this);
Expand All @@ -1426,7 +1430,12 @@ export default class Exchange {
'agent': finalAgent,
}
}, wsOptions);
this.clients[url] = new WsClient (url, onMessage, onError, onClose, onConnected, options);
this.clients[url] = new WsClient (url, onMessageWithStream, onError, onClose, onConnected, options);
const originalResolve = this.clients[url].resolve;
this.clients[url].resolve = (result, messageHash) => {
this.streamProduceHook (result, messageHash);
return originalResolve (result, messageHash);
};
}
return this.clients[url];
}
Expand Down Expand Up @@ -1781,6 +1790,25 @@ export default class Exchange {
// ------------------------------------------------------------------------
// METHODS BELOW THIS LINE ARE TRANSPILED FROM JAVASCRIPT TO PYTHON AND PHP

streamProduceHook (result, messageHash) {
const parts = messageHash.split (':');
const targetName = parts[0];
const targetNameLower = targetName.toLowerCase ();
if (targetNameLower.indexOf ('orderbook') >= 0) {
this.streamProduce ('orderbooks', result);
} else if (targetNameLower.indexOf ('ticker') >= 0) {
this.streamProduce ('tickers', result);
} else if (targetNameLower.indexOf ('order') >= 0) {
this.streamProduce ('orders', result);
} else if (targetNameLower.indexOf ('position') >= 0) {
this.streamProduce ('positions', result);
} else if (targetNameLower.indexOf ('mytrade') >= 0) {
this.streamProduce ('myTrades', result);
} else if (targetNameLower.indexOf ('trade') >= 0) {
this.streamProduce ('trades', result);
}
}

setupStream () {
const stream = this.stream;
if (this.stream !== undefined) {
Expand Down
3 changes: 0 additions & 3 deletions ts/src/pro/alpaca.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,6 @@ export default class alpaca extends alpacaRest {
const symbol = ticker['symbol'];
const messageHash = 'ticker:' + symbol;
this.tickers[symbol] = ticker;
this.streamProduce ('tickers', ticker);
client.resolve (this.tickers[symbol], messageHash);
}

Expand Down Expand Up @@ -318,7 +317,6 @@ export default class alpaca extends alpacaRest {
}
const parsed = this.parseTrade (message);
stored.append (parsed);
this.streamProduce ('myTrades', parsed);
const messageHash = 'trade' + ':' + symbol;
client.resolve (stored, messageHash);
}
Expand Down Expand Up @@ -677,7 +675,6 @@ export default class alpaca extends alpacaRest {
}

handleMessage (client: Client, message) {
this.streamProduce ('raw', message);
if (Array.isArray (message)) {
this.handleCryptoMessage (client, message);
return;
Expand Down
3 changes: 0 additions & 3 deletions ts/src/pro/ascendex.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -201,7 +201,6 @@ export default class ascendex extends ascendexRest {
}
for (let i = 0; i < trades.length; i++) {
tradesArray.append (trades[i]);
this.streamProduce ('trades', trades[i]);
}
this.trades[symbol] = tradesArray;
client.resolve (tradesArray, messageHash);
Expand Down Expand Up @@ -489,7 +488,6 @@ export default class ascendex extends ascendexRest {
}
}
const messageHash = 'balance' + ':' + type;
this.streamProduce ('balances', this.safeBalance (result));
client.resolve (this.safeBalance (result), messageHash);
}

Expand Down Expand Up @@ -742,7 +740,6 @@ export default class ascendex extends ascendexRest {
}

handleMessage (client: Client, message) {
this.streamProduce ('raw', message);
if (this.handleErrorMessage (client, message)) {
return;
}
Expand Down
40 changes: 6 additions & 34 deletions ts/src/pro/binance.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ export default class binance extends binanceRest {
return newValue;
}

streamId (type, subscriptionHash, numSubscriptions = 1) {
stream (type, subscriptionHash, numSubscriptions = 1) {
const streamBySubscriptionsHash = this.safeValue (this.options, 'streamBySubscriptionsHash', {});
let stream = this.safeString (streamBySubscriptionsHash, subscriptionHash);
if (stream === undefined) {
Expand Down Expand Up @@ -252,7 +252,7 @@ export default class binance extends binanceRest {
subParams.push (symbolHash);
}
const messageHashesLength = messageHashes.length;
const url = this.urls['api']['ws'][type] + '/' + this.streamId (type, streamHash, messageHashesLength);
const url = this.urls['api']['ws'][type] + '/' + this.stream (type, streamHash, messageHashesLength);
const requestId = this.requestId (url);
const request = {
'method': 'SUBSCRIBE',
Expand Down Expand Up @@ -321,12 +321,10 @@ export default class binance extends binanceRest {
}
}
this.orderbooks[symbol] = orderbook;
this.streamProduce ('orderbooks', orderbook);
client.resolve (orderbook, messageHash);
} catch (e) {
delete client.subscriptions[messageHash];
client.reject (e, messageHash);
this.streamProduce ('orderbooks', undefined, e);
}
}

Expand Down Expand Up @@ -418,7 +416,6 @@ export default class binance extends binanceRest {
if (conditional) {
this.handleOrderBookMessage (client, message, orderbook);
if (nonce < orderbook['nonce']) {
this.streamProduce ('orderbooks', orderbook);
client.resolve (orderbook, messageHash);
}
} else {
Expand All @@ -435,7 +432,6 @@ export default class binance extends binanceRest {
if ((U <= orderbook['nonce']) || (pu === orderbook['nonce'])) {
this.handleOrderBookMessage (client, message, orderbook);
if (nonce <= orderbook['nonce']) {
this.streamProduce ('orderbooks', orderbook);
client.resolve (orderbook, messageHash);
}
} else {
Expand All @@ -448,8 +444,6 @@ export default class binance extends binanceRest {
delete this.orderbooks[symbol];
delete client.subscriptions[messageHash];
client.reject (e, messageHash);
this.streamProduce ('orderbooks', undefined, e);
this.streamProduce ('orderbooks::' + symbol, undefined, e);
}
}
}
Expand Down Expand Up @@ -527,7 +521,7 @@ export default class binance extends binanceRest {
}
const query = this.omit (params, 'type');
const subParamsLength = subParams.length;
const url = this.urls['api']['ws'][type] + '/' + this.streamId (type, streamHash, subParamsLength);
const url = this.urls['api']['ws'][type] + '/' + this.stream (type, streamHash, subParamsLength);
const requestId = this.requestId (url);
const request = {
'method': 'SUBSCRIBE',
Expand Down Expand Up @@ -741,7 +735,6 @@ export default class binance extends binanceRest {
tradesArray = new ArrayCache (limit);
}
tradesArray.append (trade);
this.streamProduce ('trades', trade);
this.trades[symbol] = tradesArray;
client.resolve (tradesArray, messageHash);
}
Expand Down Expand Up @@ -775,7 +768,7 @@ export default class binance extends binanceRest {
if (market['contract']) {
type = market['linear'] ? 'future' : 'delivery';
}
const url = this.urls['api']['ws'][type] + '/' + this.streamId (type, messageHash);
const url = this.urls['api']['ws'][type] + '/' + this.stream (type, messageHash);
const requestId = this.requestId (url);
const request = {
'method': 'SUBSCRIBE',
Expand Down Expand Up @@ -857,8 +850,6 @@ export default class binance extends binanceRest {
this.ohlcvs[symbol][timeframe] = stored;
}
stored.append (parsed);
this.streamProduce ('ohlcv', parsed);
this.streamProduce ('ohlcv::' + symbol + '::' + timeframe, parsed);
client.resolve (stored, messageHash);
}

Expand Down Expand Up @@ -974,7 +965,7 @@ export default class binance extends binanceRest {
name = this.safeString (params, 'name', name);
params = this.omit (params, 'name');
const messageHash = marketId + '@' + name;
const url = this.urls['api']['ws'][type] + '/' + this.streamId (type, messageHash);
const url = this.urls['api']['ws'][type] + '/' + this.stream (type, messageHash);
const requestId = this.requestId (url);
const request = {
'method': 'SUBSCRIBE',
Expand Down Expand Up @@ -1036,7 +1027,7 @@ export default class binance extends binanceRest {
'!' + name + '@arr',
];
}
const url = this.urls['api']['ws'][type] + '/' + this.streamId (type, messageHash);
const url = this.urls['api']['ws'][type] + '/' + this.stream (type, messageHash);
const requestId = this.requestId (url);
const request = {
'method': 'SUBSCRIBE',
Expand Down Expand Up @@ -1181,7 +1172,6 @@ export default class binance extends binanceRest {
const symbol = result['symbol'];
this.tickers[symbol] = result;
client.resolve (result, messageHash);
this.streamProduce ('tickers', result);
if (event === 'bookTicker') {
// watch bookTickers
client.resolve (result, '!' + 'bookTicker@arr');
Expand Down Expand Up @@ -1214,7 +1204,6 @@ export default class binance extends binanceRest {
const symbol = result['symbol'];
this.tickers[symbol] = result;
newTickers[symbol] = result;
this.streamProduce ('tickers', result);
}
const messageHashes = this.findMessageHashes (client, 'tickers::');
for (let i = 0; i < messageHashes.length; i++) {
Expand Down Expand Up @@ -1363,7 +1352,6 @@ export default class binance extends binanceRest {
for (let i = 0; i < messageHashes.length; i++) {
const messageHash = messageHashes[i];
client.reject (error, messageHash);
this.streamProduce ('errors', undefined, error);
}
this.options[type] = this.extend (options, {
'listenKey': undefined,
Expand Down Expand Up @@ -1420,7 +1408,6 @@ export default class binance extends binanceRest {
// don't remove the future from the .futures cache
const future = client.futures[messageHash];
future.resolve ();
this.streamProduce ('balances', this.balance[type]);
client.resolve (this.balance[type], type + ':balance');
}

Expand Down Expand Up @@ -1504,7 +1491,6 @@ export default class binance extends binanceRest {
const messageHash = this.safeString (message, 'id');
const result = this.safeValue (message, 'result', {});
const parsedBalances = this.parseBalance (result);
this.streamProduce ('balances', parsedBalances);
client.resolve (parsedBalances, messageHash);
}

Expand Down Expand Up @@ -1651,7 +1637,6 @@ export default class binance extends binanceRest {
this.balance[accountType]['timestamp'] = timestamp;
this.balance[accountType]['datetime'] = this.iso8601 (timestamp);
this.balance[accountType] = this.safeBalance (this.balance[accountType]);
this.streamProduce ('balances', this.balance[accountType]);
client.resolve (this.balance[accountType], messageHash);
}

Expand Down Expand Up @@ -1774,7 +1759,6 @@ export default class binance extends binanceRest {
const messageHash = this.safeString (message, 'id');
const result = this.safeValue (message, 'result', {});
const order = this.parseOrder (result);
this.streamProduce ('orders', order);
client.resolve (order, messageHash);
}

Expand Down Expand Up @@ -2546,7 +2530,6 @@ export default class binance extends binanceRest {
const contracts = this.safeNumber (position, 'contracts', 0);
if (contracts > 0) {
cache.append (position);
this.streamProduce ('positions', position);
}
}
// don't remove the future from the .futures cache
Expand Down Expand Up @@ -2605,7 +2588,6 @@ export default class binance extends binanceRest {
position['datetime'] = this.iso8601 (timestamp);
newPositions.push (position);
cache.append (position);
this.streamProduce ('positions', position);
}
const messageHashes = this.findMessageHashes (client, accountType + ':positions::');
for (let i = 0; i < messageHashes.length; i++) {
Expand Down Expand Up @@ -2819,9 +2801,6 @@ export default class binance extends binanceRest {
const messageHash = this.safeString (message, 'id');
const result = this.safeValue (message, 'result', []);
const trades = this.parseTrades (result);
for (let i = 0; i < trades.length; i++) {
this.streamProduce ('myTrades', trades[i]);
}
client.resolve (trades, messageHash);
}

Expand Down Expand Up @@ -2940,7 +2919,6 @@ export default class binance extends binanceRest {
}
const myTrades = this.myTrades;
myTrades.append (trade);
this.streamProduce ('myTrades');
client.resolve (this.myTrades, messageHash);
const messageHashSymbol = messageHash + ':' + symbol;
client.resolve (this.myTrades, messageHashSymbol);
Expand Down Expand Up @@ -2978,7 +2956,6 @@ export default class binance extends binanceRest {
cachedOrders.append (parsed);
const messageHash = 'orders';
const symbolSpecificMessageHash = 'orders:' + symbol;
this.streamProduce ('orders', parsed);
client.resolve (cachedOrders, messageHash);
client.resolve (cachedOrders, symbolSpecificMessageHash);
}
Expand Down Expand Up @@ -3010,32 +2987,27 @@ export default class binance extends binanceRest {
rejected = true;
// private endpoint uses id as messageHash
client.reject (e, id);
this.streamProduce ('errors', undefined, error);
// public endpoint stores messageHash in subscriptios
const subscriptionKeys = Object.keys (client.subscriptions);
for (let i = 0; i < subscriptionKeys.length; i++) {
const subscriptionHash = subscriptionKeys[i];
const subscriptionId = this.safeString (client.subscriptions[subscriptionHash], 'id');
if (id === subscriptionId) {
client.reject (e, subscriptionHash);
this.streamProduce ('errors', undefined, error);
}
}
}
if (!rejected) {
client.reject (message, id);
this.streamProduce ('errors', undefined, error);
}
// reset connection if 5xx error
if (this.safeString (code, 0) === '5') {
client.reset (message);
this.streamProduce ('errors', undefined, message);
}
}

handleMessage (client: Client, message) {
// handle WebSocketAPI
this.streamProduce ('raw', message);
const status = this.safeString (message, 'status');
const error = this.safeValue (message, 'error');
if ((error !== undefined) || (status !== undefined && status !== '200')) {
Expand Down
3 changes: 0 additions & 3 deletions ts/src/pro/bingx.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -170,7 +170,6 @@ export default class bingx extends bingxRest {
const ticker = this.parseWsTicker (data, market);
this.tickers[symbol] = ticker;
const messageHash = market['id'] + '@ticker';
this.streamProduce ('tickers', ticker);
client.resolve (ticker, messageHash);
}

Expand Down Expand Up @@ -342,7 +341,6 @@ export default class bingx extends bingxRest {
}
for (let j = 0; j < trades.length; j++) {
stored.append (trades[j]);
this.streamProduce ('trades', trades[j]);
}
client.resolve (stored, messageHash);
}
Expand Down Expand Up @@ -1084,7 +1082,6 @@ export default class bingx extends bingxRest {
}

handleMessage (client: Client, message) {
this.streamProduce ('raw', message);
if (!this.handleErrorMessage (client, message)) {
return;
}
Expand Down
4 changes: 0 additions & 4 deletions ts/src/pro/bitfinex.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,6 @@ export default class bitfinex extends bitfinexRest {
const trades = this.parseTrades (data, market);
for (let i = 0; i < trades.length; i++) {
stored.append (trades[i]);
this.streamProduce ('trades', trades[i]);
}
} else {
const second = this.safeString (message, 1);
Expand All @@ -137,7 +136,6 @@ export default class bitfinex extends bitfinexRest {
}
const trade = this.parseTrade (message, market);
stored.append (trade);
this.streamProduce ('trades', trade);
}
client.resolve (stored, messageHash);
}
Expand Down Expand Up @@ -251,7 +249,6 @@ export default class bitfinex extends bitfinexRest {
'info': message,
};
this.tickers[symbol] = result;
this.streamProduce ('tickers', result);
client.resolve (result, messageHash);
}

Expand Down Expand Up @@ -628,7 +625,6 @@ export default class bitfinex extends bitfinexRest {
}

handleMessage (client: Client, message) {
this.streamProduce ('raw', message);
if (Array.isArray (message)) {
const channelId = this.safeString (message, 0);
//
Expand Down
Loading