Skip to content

Commit

Permalink
Rename RealtimeChannel#onMessage
Browse files Browse the repository at this point in the history 
As in ab22a56.
  • Loading branch information
@lawrence-forooghian
lawrence-forooghian committed Jun 5, 2023
1 parent ab22a56 commit a407328
Show file tree
Hide file tree
Showing 5 changed files with 50 additions and 46 deletions.
2 changes: 1 addition & 1 deletion src/common/lib/client/realtime.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ class Channels extends EventEmitter {
);
return;
}
channel.onMessage(msg);
channel.processMessage(msg);
}

/* called when a transport becomes connected; reattempt attach/detach
Expand Down
18 changes: 11 additions & 7 deletions src/common/lib/client/realtimechannel.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -582,7 +582,7 @@ class RealtimeChannel extends Channel {
this.sendMessage(msg, callback);
}

onMessage(message: ProtocolMessage): void {
processMessage(message: ProtocolMessage): void {
if (
message.action === actions.ATTACHED ||
message.action === actions.MESSAGE ||
Expand Down Expand Up @@ -661,7 +661,7 @@ class RealtimeChannel extends Channel {
if (!presenceMsg.timestamp) presenceMsg.timestamp = timestamp;
if (!presenceMsg.id) presenceMsg.id = id + ':' + i;
} catch (e) {
Logger.logAction(Logger.LOG_ERROR, 'RealtimeChannel.onMessage()', (e as Error).toString());
Logger.logAction(Logger.LOG_ERROR, 'RealtimeChannel.processMessage()', (e as Error).toString());
}
}
this.presence.setPresence(presence, isSync, syncChannelSerial as any);
Expand All @@ -672,7 +672,7 @@ class RealtimeChannel extends Channel {
if (this.state !== 'attached') {
Logger.logAction(
Logger.LOG_MAJOR,
'RealtimeChannel.onMessage()',
'RealtimeChannel.processMessage()',
'Message "' +
message.id +
'" skipped as this channel "' +
Expand Down Expand Up @@ -702,7 +702,7 @@ class RealtimeChannel extends Channel {
'" on this channel "' +
this.name +
'".';
Logger.logAction(Logger.LOG_ERROR, 'RealtimeChannel.onMessage()', msg);
Logger.logAction(Logger.LOG_ERROR, 'RealtimeChannel.processMessage()', msg);
this._startDecodeFailureRecovery(new ErrorInfo(msg, 40018, 400));
break;
}
Expand All @@ -713,7 +713,7 @@ class RealtimeChannel extends Channel {
Message.decode(msg, this._decodingContext);
} catch (e) {
/* decrypt failed .. the most likely cause is that we have the wrong key */
Logger.logAction(Logger.LOG_ERROR, 'RealtimeChannel.onMessage()', (e as Error).toString());
Logger.logAction(Logger.LOG_ERROR, 'RealtimeChannel.processMessage()', (e as Error).toString());
switch ((e as ErrorInfo).code) {
case 40018:
/* decode failure */
Expand Down Expand Up @@ -753,7 +753,7 @@ class RealtimeChannel extends Channel {
default:
Logger.logAction(
Logger.LOG_ERROR,
'RealtimeChannel.onMessage()',
'RealtimeChannel.processMessage()',
'Fatal protocol error: unrecognised action (' + message.action + ')'
);
this.connectionManager.abort(ConnectionErrors.unknownChannelErr());
Expand All @@ -762,7 +762,11 @@ class RealtimeChannel extends Channel {

_startDecodeFailureRecovery(reason: ErrorInfo): void {
if (!this._lastPayload.decodeFailureRecoveryInProgress) {
Logger.logAction(Logger.LOG_MAJOR, 'RealtimeChannel.onMessage()', 'Starting decode failure recovery process.');
Logger.logAction(
Logger.LOG_MAJOR,
'RealtimeChannel.processMessage()',
'Starting decode failure recovery process.'
);
this._lastPayload.decodeFailureRecoveryInProgress = true;
this._attach(true, reason, () => {
this._lastPayload.decodeFailureRecoveryInProgress = false;
Expand Down
12 changes: 6 additions & 6 deletions test/realtime/failure.test.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -327,13 +327,13 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
it('attach_timeout', function (done) {
var realtime = helper.AblyRealtime({ realtimeRequestTimeout: 2000, channelRetryTimeout: 1000 }),
channel = realtime.channels.get('failed_attach'),
originalOnMessage = channel.onMessage.bind(channel);
originalProcessMessage = channel.processMessage.bind(channel);

channel.onMessage = function (message) {
channel.processMessage = function (message) {
if (message.action === 11) {
return;
}
originalOnMessage(message);
originalProcessMessage(message);
};

realtime.connection.once('connected', function () {
Expand Down Expand Up @@ -366,17 +366,17 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
transports: [transport],
}),
channel = realtime.channels.get('failed_attach'),
originalOnMessage = channel.onMessage.bind(channel),
originalProcessMessage = channel.processMessage.bind(channel),
retryCount = 0;

var performance = isBrowser ? window.performance : require('perf_hooks').performance;

channel.onMessage = function (message) {
channel.processMessage = function (message) {
// Ignore ATTACHED messages
if (message.action === 11) {
return;
}
originalOnMessage(message);
originalProcessMessage(message);
};

realtime.connection.on('connected', function () {
Expand Down
6 changes: 3 additions & 3 deletions test/realtime/presence.test.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1732,7 +1732,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
},
function (cb) {
/* Inject an additional member locally */
channel.onMessage({
channel.processMessage({
action: 14,
id: 'messageid:0',
connectionId: 'connid',
Expand Down Expand Up @@ -1812,7 +1812,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
},
function (cb) {
/* Inject a member locally */
channel.onMessage({
channel.processMessage({
action: 14,
id: 'messageid:0',
connectionId: 'connid',
Expand Down Expand Up @@ -1846,7 +1846,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
cb();
});
/* Inject an ATTACHED with RESUMED and HAS_PRESENCE both false */
channel.onMessage(
channel.processMessage(
createPM({
action: 11,
channelSerial: channel.properties.attachSerial,
Expand Down
58 changes: 29 additions & 29 deletions test/realtime/sync.test.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
channelName = 'syncexistingset',
channel = realtime.channels.get(channelName);

channel.onMessage(
channel.processMessage(
createPM({
action: 11,
channel: channelName,
Expand All @@ -58,7 +58,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
async.series(
[
function (cb) {
channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
presence: [
Expand Down Expand Up @@ -95,7 +95,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
},
function (cb) {
/* Trigger another sync. Two has gone without so much as a `leave` message! */
channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
presence: [
Expand Down Expand Up @@ -149,7 +149,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
channelName = 'sync_member_arrives_in_middle',
channel = realtime.channels.get(channelName);

channel.onMessage(
channel.processMessage(
createPM({
action: 11,
channel: channelName,
Expand All @@ -158,7 +158,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
);

/* First sync */
channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
presence: [
Expand All @@ -173,7 +173,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
});

/* A second sync, this time in multiple parts, with a presence message in the middle */
channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
channelSerial: 'serial:cursor',
Expand All @@ -188,7 +188,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
],
});

channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
presence: [
Expand All @@ -202,7 +202,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
],
});

channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
channelSerial: 'serial:',
Expand Down Expand Up @@ -242,15 +242,15 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
channelName = 'sync_member_arrives_normally_after_came_in_sync',
channel = realtime.channels.get(channelName);

channel.onMessage(
channel.processMessage(
createPM({
action: 11,
channel: channelName,
flags: 1,
})
);

channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
channelSerial: 'serial:cursor',
Expand All @@ -265,7 +265,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
],
});

channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
presence: [
Expand All @@ -279,7 +279,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
],
});

channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
channelSerial: 'serial:',
Expand Down Expand Up @@ -319,15 +319,15 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
channelName = 'sync_member_arrives_normally_before_comes_in_sync',
channel = realtime.channels.get(channelName);

channel.onMessage(
channel.processMessage(
createPM({
action: 11,
channel: channelName,
flags: 1,
})
);

channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
channelSerial: 'serial:cursor',
Expand All @@ -342,7 +342,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
],
});

channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
presence: [
Expand All @@ -356,7 +356,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
],
});

channel.onMessage({
channel.processMessage({
action: 16,
channel: channelName,
channelSerial: 'serial:',
Expand Down Expand Up @@ -397,15 +397,15 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
channelName = 'sync_ordering',
channel = realtime.channels.get(channelName);

channel.onMessage(
channel.processMessage(
createPM({
action: 11,
channel: channelName,
})
);

/* One enters */
channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
id: 'one_connid:1',
Expand All @@ -420,7 +420,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
});

/* An earlier leave from one (should be ignored) */
channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
connectionId: 'one_connid',
Expand All @@ -435,7 +435,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
});

/* One adds some data in a newer msgSerial */
channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
connectionId: 'one_connid',
Expand All @@ -451,7 +451,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
});

/* Two enters */
channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
connectionId: 'two_connid',
Expand All @@ -466,7 +466,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
});

/* Two updates twice in the same message */
channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
connectionId: 'two_connid',
Expand All @@ -487,7 +487,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
});

/* Three enters */
channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
connectionId: 'three_connid',
Expand All @@ -503,7 +503,7 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async

/* Synthesized leave for three (with earlier msgSerial, incompatible id,
* and later timestamp) */
channel.onMessage({
channel.processMessage({
action: 14,
channel: channelName,
connectionId: 'synthesized',
Expand Down Expand Up @@ -581,13 +581,13 @@ define(['ably', 'shared_helper', 'async', 'chai'], function (Ably, helper, async
);
},
function (cb) {
var originalOnMessage = syncerChannel.onMessage;
syncerChannel.onMessage = function (message) {
originalOnMessage.apply(this, arguments);
var originalProcessMessage = syncerChannel.processMessage;
syncerChannel.processMessage = function (message) {
originalProcessMessage.apply(this, arguments);
/* Inject an additional presence message after the first sync */
if (message.action === 16) {
syncerChannel.onMessage = originalOnMessage;
syncerChannel.onMessage({
syncerChannel.processMessage = originalProcessMessage;
syncerChannel.processMessage({
action: 14,
id: 'messageid:0',
connectionId: 'connid',
Expand Down

0 comments on commit a407328

Please sign in to comment.