Skip to content

Commit 40cfc6c

Browse files
sponzillosponzillo
committed
## v0.2.21
- amqplib ^0.7.1 => ^0.8.0 - node 12 => 16.17.1
1 parent 1b9bb41 commit 40cfc6c

File tree

9 files changed

+1382
-1264
lines changed

9 files changed

+1382
-1264
lines changed

‎CHANGELOG.md‎

Lines changed: 31 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,23 @@
1+
2+
**npm @chat21/chat21-server@0.2.20**
3+
4+
available on:
5+
▶️ https://www.npmjs.com/package/@chat21/chat21-server
6+
7+
## v0.2.21 - online
8+
- amqplib ^0.7.1 => ^0.8.0
9+
- node 12 => 16.17.1
10+
11+
## v0.2.20
12+
- added process.exit(0) on "[AMQP] channel error". It lets the server to silently restart on blocking AMQP errors.
13+
14+
## v0.2.19
15+
- ack in sendMessageToGroupMembers() sent immediately.
16+
- added group_id as memebr of inlineGroup.
17+
18+
## v0.2.18
19+
- Added inlineGroup management. You can create on the fly group just sending a message with the "group.members" attribute.
20+
121
## v0.2.17
222
- "ack" management improvements
323

@@ -53,10 +73,17 @@ to only enable "messages" queue.
5373
## v0.1.13 npm online
5474

5575
- bugfix:
56-
this: if (inbox_of === outgoing_message.sender) {
57-
became: if (inbox_of === group.uid) { // choosing one member, the group ("volatile" member), for the "status=SENT", used by the "message-sent" webhook
76+
this:
77+
if (inbox_of === outgoing_message.sender) {
78+
became:
79+
if (inbox_of === group.uid) {
80+
logger.debug("inbox_of === outgoing_message.sender. status=SENT system YES?", inbox_of);
81+
outgoing_message.status = MessageConstants.CHAT_MESSAGE_STATUS_CODE.SENT;
82+
}
83+
// choosing one member, the group ("volatile" member), for the "status=SENT", used by the "message-sent" webhook
84+
// achived changing the delivered status to SENT "on-the-fly" when I deliver the message to the group id. This
85+
will trigger the webhookSentOrDelivered to "Sent" only
5886

5987
If "system" sends info messages and he is not member of the group, webhooks are never called.
6088
The "message-sent" webhook is called only once: when, iterating all the members, the selected one is the same as the group.
61-
This because the "message-sent" must be called only once per message. The "sender" can't be used, because the "sender" not always
62-
is a group's member (ex. info messages by system while system is not always a member of the group).
89+
This because the "message-sent" must be called only once per message. The "sender" can't be used, because the "sender" not always is a group's member (ex. info messages by system while system is not always a member of the group).

‎benchmarks/performance_test.js‎

Lines changed: 40 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -21,19 +21,33 @@ const { Console } = require('console');
2121

2222
// const APPID = 'tilechat';
2323

24+
// CONSOLE.NATIVE
2425
let config = {
2526
EXPECTED_AVG_DIRECT_MESSAGE_DELAY: 160,
2627
EXPECTED_AVG_GROUP_MESSAGE_DELAY: 160,
27-
REQS_PER_SECOND: 300,
28+
REQS_PER_SECOND: 100,
2829
MAX_SECONDS: 10,
2930
CONCURRENCY: 1, // 2
3031
API_SERVER_HOST: 'localhost',
3132
API_SERVER_PORT: 8004,
32-
MQTT_ENDPOINT: 'ws://localhost:15675/ws',
33-
API_ENDPOINT: 'http://localhost:8004/api',
33+
MQTT_ENDPOINT: 'wss://console.native.tiledesk.com:15675/ws',
34+
API_ENDPOINT: 'https://console.native.tiledesk.com/chatapi/api',
3435
APPID: 'tilechat'
3536
}
3637

38+
// let config = {
39+
// EXPECTED_AVG_DIRECT_MESSAGE_DELAY: 160,
40+
// EXPECTED_AVG_GROUP_MESSAGE_DELAY: 160,
41+
// REQS_PER_SECOND: 100,
42+
// MAX_SECONDS: 10,
43+
// CONCURRENCY: 1, // 2
44+
// //API_SERVER_HOST: 'localhost',
45+
// API_SERVER_PORT: 8004,
46+
// MQTT_ENDPOINT: 'ws://localhost:15675/ws',
47+
// API_ENDPOINT: 'http://localhost:8004/api',
48+
// APPID: 'tilechat'
49+
// }
50+
3751
const user1 = {
3852
userid: 'USER1',
3953
fullname: 'User 1',
@@ -54,7 +68,8 @@ let chatClient1 = new Chat21Client(
5468
{
5569
appId: config.APPID,
5670
MQTTendpoint: config.MQTT_ENDPOINT,
57-
APIendpoint: config.API_ENDPOINT
71+
APIendpoint: config.API_ENDPOINT,
72+
log: true
5873
});
5974

6075
let chatClient2 = new Chat21Client(
@@ -86,18 +101,22 @@ let group_name; // got in before()
86101
describe("Performance Test", function() {
87102
before(function(done) {
88103
chatClient1.connect(user1.userid, user1.token, () => {
89-
// console.log("chatClient1 Connected...");
104+
console.log("chatClient1 Connected...");
90105
chatClient2.connect(user2.userid, user2.token, async () => {
91-
// console.log("chatClient2 Connected...");
106+
console.log("chatClient2 Connected...");
92107
group_id = "group-" + uuidv4().replace("-", "");
93108
group_name = "benchmarks group " + group_id;
94109
const group_members = {}
95110
group_members[user2.userid] = 1;
111+
let total_ = 0
112+
const start_ = Date.now();
96113
chatClient1.groupCreate(
97114
group_name,
98115
group_id,
99116
group_members,
100117
(err, result) => {
118+
total_ = Date.now() - start_
119+
console.log("TOTAL GROUP CREATION TIME", total_)
101120
assert(err == null);
102121
assert(result != null);
103122
assert(result.success == true);
@@ -145,13 +164,13 @@ describe("Performance Test", function() {
145164
let total_iterations = config.REQS_PER_SECOND * config.MAX_SECONDS;
146165
let test_start_time = Date.now();
147166
let current = 0;
148-
console.log("Direct - message average latency is expected to be <", config.EXPECTED_AVG_DIRECT_MESSAGE_DELAY + "ms");
149-
console.log("Direct - MESSAGES/SEC =", config.REQS_PER_SECOND * config.CONCURRENCY);
150-
console.log("Direct - MESSAGES/SEC/VU =", config.REQS_PER_SECOND);
151-
console.log("Direct - TEST DURATION (s) =", config.MAX_SECONDS);
152-
console.log("Direct - CONCURRENCY (#VUs) =", config.CONCURRENCY);
153-
console.log("Direct - DELAY BETWEEN MESSAGES (ms) =", delay);
154-
console.log("Direct - TOTAL ITERATIONS =", total_iterations);
167+
console.log("Direct - Expected message average latency to be <", config.EXPECTED_AVG_DIRECT_MESSAGE_DELAY + "ms");
168+
console.log("Direct - Expected MESSAGES/SEC =", config.REQS_PER_SECOND * config.CONCURRENCY);
169+
console.log("Direct - Expected MESSAGES/SEC/VU =", config.REQS_PER_SECOND);
170+
console.log("Direct - Expected TEST DURATION (s) =", config.MAX_SECONDS);
171+
console.log("Direct - Expected CONCURRENCY (#VUs) =", config.CONCURRENCY);
172+
console.log("Direct - Expected DELAY BETWEEN MESSAGES (ms) =", delay);
173+
console.log("Direct - Expected TOTAL ITERATIONS =", total_iterations);
155174
console.log("Direct - Running benchmark...");
156175

157176
for (let i = 0; i < total_iterations; i++) {
@@ -175,7 +194,7 @@ describe("Performance Test", function() {
175194

176195
function endCallback(latency) {
177196
console.log("\n\n********* Direct - Benchmark results *********");
178-
console.log("Direct - Final latency:", latency.meanLatencyMs);
197+
console.log("Direct - Message mean latency:", latency.meanLatencyMs);
179198
let test_duration = Math.round(current / 1000)
180199
console.log("Direct - Test duration:", test_duration + " seconds" + " (" + current + ") ms");
181200
let mesg_sec = Math.round(latency.totalMessages / test_duration)
@@ -205,13 +224,13 @@ describe("Performance Test", function() {
205224
let total_iterations = config.REQS_PER_SECOND * config.MAX_SECONDS;
206225
let test_start_time = Date.now();
207226
let current = 0;
208-
console.log("Group - message average latency is expected to be <", config.EXPECTED_AVG_DIRECT_MESSAGE_DELAY + "ms");
209-
console.log("Group - CONCURRENCY (#VIRTUAL USERs aka VUs) =", config.CONCURRENCY);
210-
console.log("Group - MESSAGES/SEC =", config.REQS_PER_SECOND * config.CONCURRENCY);
211-
console.log("Group - MESSAGES/SEC/VU =", config.REQS_PER_SECOND);
212-
console.log("Group - TEST DURATION (s) =", config.MAX_SECONDS);
213-
console.log("Group - DELAY BETWEEN MESSAGES (ms) =", delay);
214-
console.log("Group - TOTAL ITERATIONS =", total_iterations);
227+
console.log("Group - Expected message average latency to be <", config.EXPECTED_AVG_DIRECT_MESSAGE_DELAY + "ms");
228+
console.log("Group - Expected CONCURRENCY (#VIRTUAL USERs aka VUs) =", config.CONCURRENCY);
229+
console.log("Group - Expected MESSAGES/SEC =", config.REQS_PER_SECOND * config.CONCURRENCY);
230+
console.log("Group - Expected MESSAGES/SEC/VU =", config.REQS_PER_SECOND);
231+
console.log("Group - Expected TEST DURATION (s) =", config.MAX_SECONDS);
232+
console.log("Group - Expected DELAY BETWEEN MESSAGES (ms) =", delay);
233+
console.log("Group - Expected TOTAL ITERATIONS =", total_iterations);
215234
console.log("Group - Running benchmark...");
216235
for (let i = 0; i < total_iterations; i++) {
217236
// console.log("GROUP i:", i)

‎mqttclient/chat21client.js‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -127,7 +127,7 @@ class Chat21Client {
127127
// callback - function (err)
128128
// console.log("recipient_id:", recipient_id)
129129
let dest_topic = `apps/${this.appid}/outgoing/users/${this.user_id}/messages/${recipient_id}/outgoing`
130-
// console.log("dest_topic:", dest_topic)
130+
console.log("dest_topic:", dest_topic)
131131
// let outgoing_message = {
132132
// text: text,
133133
// type: type,
@@ -537,7 +537,9 @@ class Chat21Client {
537537
this.subscribeToMyConversations()
538538
// no more then one "on_message" handler, thanks.
539539
this.on_message_handler = this.client.on('message', (topic, message) => {
540-
// console.log("topic:" + topic + "\nmessage payload:" + message)
540+
if (this.log) {
541+
console.log("topic:" + topic + "\nmessage payload:" + message)
542+
}
541543
const _topic = this.parseTopic(topic)
542544
if (!_topic) {
543545
if (this.log) {

‎observer.js‎

Lines changed: 70 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,7 @@ function startPublisher() {
165165
if (closeOnErr(err)) return;
166166
ch.on("error", function (err) {
167167
logger.error("[AMQP] channel error", err);
168+
process.exit(0);
168169
});
169170
ch.on("close", function () {
170171
logger.debug("[AMQP] channel closed");
@@ -216,6 +217,7 @@ function startWorker() {
216217
if (closeOnErr(err)) return;
217218
ch.on("error", function (err) {
218219
logger.error("[AMQP] channel error", err);
220+
process.exit(0);
219221
});
220222
ch.on("close", function () {
221223
logger.debug("[AMQP] channel closed");
@@ -407,11 +409,21 @@ function process_outgoing(topic, message_string, callback) {
407409
});
408410
}
409411
else {
410-
logger.debug("Group message.");
411412
const group_id = recipient_id
413+
if (outgoing_message.group) {
414+
logger.debug("Inline Group message.", outgoing_message);
415+
let inline_group = outgoing_message.group;
416+
inline_group.uid = group_id;
417+
inline_group.members[group_id] = 1
418+
inline_group.members[sender_id] = 1
419+
console.log("...inline_group:", inline_group);
420+
sendMessageToGroupMembers(outgoing_message, inline_group, app_id, (ack) => {
421+
callback(ack);
422+
});
423+
return;
424+
}
412425
// chatdb.getGroup(group_id, function(err, group) { // REDIS?
413426
getGroup(group_id, function(err, group) {
414-
// logger.debug("group found!", group)
415427
if (!group) { // created only to temporary store group-messages in group-timeline
416428
// TODO: 1. create group (on-the-fly), 2. remove this code, 3. continue as if the group exists.
417429
logger.debug("group doesn't exist! Sending anyway to group timeline...");
@@ -421,65 +433,67 @@ function process_outgoing(topic, message_string, callback) {
421433
members: {
422434
}
423435
}
424-
group.members[me] = 1
436+
group.members[sender_id] = 1
425437
}
426-
// if (!group.members[me]) {
427-
// logger.debug(me + " can't write to this group")
428-
// callback(true)
429-
// return
430-
// }
431438

432439
// Adding the group as a "volatile" member, so we easily get a copy of
433440
// all the group messages in timelineOf: group.uid
434441
group.members[group.uid] = 1
435-
// logger.debug("Writing to group:", group)
436-
let count = 0;
437-
logger.debug("group members", group.members);
438-
let max = Object.keys(group.members).length;
439-
let error_encoutered = false;
440-
for (let [member_id, value] of Object.entries(group.members)) {
441-
const inbox_of = member_id;
442-
const convers_with = recipient_id;
443-
logger.debug("inbox_of: "+ inbox_of);
444-
logger.debug("convers_with: " + convers_with);
445-
logger.debug("sending group outgoing message to member", member_id);
446-
// if (inbox_of === outgoing_message.sender) {
447-
if (inbox_of === group.uid) { // choosing one member, the group ("volatile" member), for the "status=SENT", used by the "message-sent" webhook
448-
logger.debug("inbox_of === outgoing_message.sender. status=SENT system YES?", inbox_of);
449-
outgoing_message.status = MessageConstants.CHAT_MESSAGE_STATUS_CODE.SENT;
450-
}
451-
else if (outgoing_message.attributes && outgoing_message.attributes.hiddenFor && outgoing_message.attributes.hiddenFor === inbox_of) {
452-
logger.debug('sendGroupMessageToMembersTimeline skip message for ' + outgoing_message.attributes.hiddenFor);
453-
break;
442+
sendMessageToGroupMembers(outgoing_message, group, app_id, (ack) => {
443+
callback(ack);
444+
});
445+
})
446+
}
447+
}
448+
449+
function sendMessageToGroupMembers(outgoing_message, group, app_id, callback) {
450+
// logger.debug("Writing to group:", group)
451+
let count = 0;
452+
logger.debug("group members", group.members);
453+
let max = Object.keys(group.members).length;
454+
let error_encoutered = false;
455+
for (let [member_id, value] of Object.entries(group.members)) {
456+
const inbox_of = member_id;
457+
const convers_with = group.uid;
458+
logger.debug("inbox_of: "+ inbox_of);
459+
logger.debug("convers_with: " + convers_with);
460+
logger.debug("sending group outgoing message to member", member_id);
461+
// if (inbox_of === outgoing_message.sender) {
462+
if (inbox_of === group.uid) { // choosing one member, the group ("volatile" member), for the "status=SENT", used by the "message-sent" webhook
463+
logger.debug("inbox_of === outgoing_message.sender. status=SENT system YES?", inbox_of);
464+
outgoing_message.status = MessageConstants.CHAT_MESSAGE_STATUS_CODE.SENT;
465+
}
466+
else if (outgoing_message.attributes && outgoing_message.attributes.hiddenFor && outgoing_message.attributes.hiddenFor === inbox_of) {
467+
logger.debug('sendGroupMessageToMembersTimeline skip message for ' + outgoing_message.attributes.hiddenFor);
468+
break;
469+
}
470+
else {
471+
logger.debug("inbox_of != outgoing_message.sender. status=DELIVERED no system, is:", inbox_of);
472+
outgoing_message.status = MessageConstants.CHAT_MESSAGE_STATUS_CODE.DELIVERED;
473+
}
474+
console.log("delivering group message with status...", outgoing_message.status, " to:", inbox_of);
475+
deliverMessage(outgoing_message, app_id, inbox_of, convers_with, function(ok) {
476+
logger.debug("GROUP MESSAGE DELIVERED?", ok)
477+
count++;
478+
logger.debug("Sent Counting:", count);
479+
logger.debug("Max:", max);
480+
if (!ok) {
481+
logger.debug("Error sending message to group " + group.uid);
482+
error_encoutered = true
483+
}
484+
if (count == max) {
485+
if (error_encoutered) {
486+
logger.error("ERROR SENDING MESSAGE TO GROUP!");
487+
//callback(false)
454488
}
455489
else {
456-
logger.debug("inbox_of != outgoing_message.sender. status=DELIVERED no system, is:", inbox_of);
457-
outgoing_message.status = MessageConstants.CHAT_MESSAGE_STATUS_CODE.DELIVERED;
490+
logger.log("ALL OK! MESSAGE SENT TO GROUP! ACK!");
491+
//callback(true);
458492
}
459-
logger.debug("delivering group message with status...", outgoing_message.status, " to:", inbox_of);
460-
deliverMessage(outgoing_message, app_id, inbox_of, convers_with, function(ok) {
461-
logger.debug("GROUP MESSAGE DELIVERED?", ok)
462-
count++;
463-
logger.debug("Sent Counting:", count);
464-
logger.debug("Max:", max);
465-
if (!ok) {
466-
logger.debug("Error sending message to group " + group.uid);
467-
error_encoutered = true
468-
}
469-
if (count == max) {
470-
if (error_encoutered) {
471-
logger.error("ERROR SENDING MESSAGE TO GROUP!");
472-
callback(false)
473-
}
474-
else {
475-
logger.log("ALL OK! MESSAGE SENT TO GROUP! ACK!");
476-
callback(true);
477-
}
478-
}
479-
})
480-
} // end for
481-
}) // end getGroup
482-
}
493+
}
494+
})
495+
} // end for
496+
callback(true);
483497
}
484498

485499
let groups = {};
@@ -755,6 +769,7 @@ function process_update(topic, message_string, callback) {
755769
// }
756770

757771
// TODO: MOVE TO A PERSIST_UPDATED TOPIC/QUEUE...
772+
// TODO, BETTER: USE _WEBHOOK WITH MESSAGE-STATUS-UPDATED TO SAVE THE MESSAGE
758773
logger.debug(">>> ON DISK... WITH A STATUS ON MY MESSAGE-UPDATE TOPIC", topic, "WITH PATCH", my_message_patch)
759774
chatdb.saveOrUpdateMessage(my_message_patch, function(err, msg) {
760775
// logger.debug(">>> MESSAGE ON TOPIC", topic, "UPDATED!")
@@ -795,6 +810,7 @@ function process_update(topic, message_string, callback) {
795810
patch.conversWith = convers_with
796811
logger.debug(">>> ON DISK... CONVERSATION TOPIC " + topic + " WITH PATCH " + patch)
797812
logger.debug("Updating conversation 2.")
813+
// BETTER: ACK, THEN WEBHOOK CONVERSATION-SAVE
798814
chatdb.saveOrUpdateConversation(patch, function(err, doc) {
799815
logger.debug(">>> CONVERSATION ON TOPIC:", topic, "UPDATED?")
800816
if (err) {
@@ -845,6 +861,7 @@ function process_archive(topic, payload, callback) {
845861
}
846862
logger.debug("NOTIFY VIA WEBHOOK ON SAVE TOPIC "+ topic)
847863
if (webhook_enabled) {
864+
// BETTER: WEBHOOK CONVERSATION-SAVE WITH archived=true
848865
webhooks.WHnotifyConversationArchived(conversation_archive_patch, topic, (err) => {
849866
if (err) {
850867
logger.error("Webhook notified with err:"+ err)

0 commit comments

Comments
 (0)