Skip to content

Commit

Permalink
## v0.2.11
Browse files Browse the repository at this point in the history 
- archive-conversation payload now publishes on MQTT the full conversation data, not only the conversation patch
- improved test #17 to check full archived conversation payload
- updated user auth tokens in performance test to match new chat auth path
- added conversationDetail(appid, timelineOf, conversWith, archived, callback)
- removed getConversation (unused method)
  • Loading branch information
@sponzillo
sponzillo committed Feb 12, 2022
1 parent 7ddc9aa commit cdfa0eb
Show file tree
Hide file tree
Showing 6 changed files with 69 additions and 36 deletions.
3 changes: 3 additions & 0 deletions CHANGELOG.md
View file
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@

## v0.2.11
- archive-conversation payload now publishes on MQTT the full conversation data, not only the conversation patch

## v0.2.10
- added test #17 - conversation/archivedConversation detail

Expand Down
26 changes: 12 additions & 14 deletions benchmarks/performance_test.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -39,23 +39,15 @@ const user1 = {
fullname: 'User 1',
firstname: 'User',
lastname: '1',
token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOiI2NzI5NDc4ZS1mOWIwLTRiODctYjNhYS03ZjU1OWExNzc5YjIiLCJzdWIiOiJVU0VSMSIsInNjb3BlIjpbInJhYmJpdG1xLnJlYWQ6Ki8qL2FwcHMudGlsZWNoYXQudXNlcnMuVVNFUjEuKiIsInJhYmJpdG1xLndyaXRlOiovKi9hcHBzLnRpbGVjaGF0LnVzZXJzLlVTRVIxLioiLCJyYWJiaXRtcS5jb25maWd1cmU6Ki8qLyoiXSwiY2xpZW50X2lkIjoiVVNFUjEiLCJjaWQiOiJVU0VSMSIsImF6cCI6IlVTRVIxIiwidXNlcl9pZCI6IlVTRVIxIiwiYXBwX2lkIjoidGlsZWNoYXQiLCJpYXQiOjE2MjM3Njc1MjAsImV4cCI6MTkzNDgwNzUyMCwiYXVkIjpbInJhYmJpdG1xIiwiVVNFUjEiXSwia2lkIjoidGlsZWRlc2sta2V5IiwidGlsZWRlc2tfYXBpX3JvbGVzIjoidXNlciJ9.r-GBXo1fIUtl1QjOkXxcRaenVNQBElRkus3omh9YtjQ'
token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOiI2OGFkODJjYi1lODE2LTRkYWEtYjljYi0wM2NiZmFjMDY1OGQiLCJzdWIiOiJVU0VSMSIsInNjb3BlIjpbInJhYmJpdG1xLnJlYWQ6Ki8qL2FwcHMudGlsZWNoYXQudXNlcnMuVVNFUjEuKiIsInJhYmJpdG1xLndyaXRlOiovKi9hcHBzLnRpbGVjaGF0LnVzZXJzLlVTRVIxLioiLCJyYWJiaXRtcS53cml0ZToqLyovYXBwcy50aWxlY2hhdC5vdXRnb2luZy51c2Vycy5VU0VSMS4qIiwicmFiYml0bXEuY29uZmlndXJlOiovKi8qIl0sImNsaWVudF9pZCI6IlVTRVIxIiwiY2lkIjoiVVNFUjEiLCJhenAiOiJVU0VSMSIsInVzZXJfaWQiOiJVU0VSMSIsImFwcF9pZCI6InRpbGVjaGF0IiwiaWF0IjoxNjQ0Njc1NzcxLCJleHAiOjE5NTU3MTU3NzEsImF1ZCI6WyJyYWJiaXRtcSIsIlVTRVIxIl0sImtpZCI6InRpbGVkZXNrLWtleSIsInRpbGVkZXNrX2FwaV9yb2xlcyI6InVzZXIifQ.CrvQLL3DMydcRyLSyfyJBSdyG-HKDj5Pd8kA1UIPjQA'
};

const user2 = {
userid: 'USER2',
fullname: 'User 2',
firstname: 'User',
lastname: '2',
token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOiI0MmEwMjc2ZC1lODUzLTQ5YjMtOTU4ZS0xODBkMjFjZGZjNWMiLCJzdWIiOiJVU0VSMiIsInNjb3BlIjpbInJhYmJpdG1xLnJlYWQ6Ki8qL2FwcHMudGlsZWNoYXQudXNlcnMuVVNFUjIuKiIsInJhYmJpdG1xLndyaXRlOiovKi9hcHBzLnRpbGVjaGF0LnVzZXJzLlVTRVIyLioiLCJyYWJiaXRtcS5jb25maWd1cmU6Ki8qLyoiXSwiY2xpZW50X2lkIjoiVVNFUjIiLCJjaWQiOiJVU0VSMiIsImF6cCI6IlVTRVIyIiwidXNlcl9pZCI6IlVTRVIyIiwiYXBwX2lkIjoidGlsZWNoYXQiLCJpYXQiOjE2MjM3Njc1MjAsImV4cCI6MTkzNDgwNzUyMCwiYXVkIjpbInJhYmJpdG1xIiwiVVNFUjIiXSwia2lkIjoidGlsZWRlc2sta2V5IiwidGlsZWRlc2tfYXBpX3JvbGVzIjoidXNlciJ9.Zkbr3e9MfGGDKRdVUyG4330LxeNaKYS0y3upPtS4Wgg'
};

const user3 = {
userid: 'USER3',
fullname: 'User 3',
firstname: 'User',
lastname: '3',
token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOiJlMmI2Y2RhMi0yNjhmLTQxZDMtYjBjYy1kZWNjN2I0M2UwMjEiLCJzdWIiOiJVU0VSMyIsInNjb3BlIjpbInJhYmJpdG1xLnJlYWQ6Ki8qL2FwcHMudGlsZWNoYXQudXNlcnMuVVNFUjMuKiIsInJhYmJpdG1xLndyaXRlOiovKi9hcHBzLnRpbGVjaGF0LnVzZXJzLlVTRVIzLioiLCJyYWJiaXRtcS5jb25maWd1cmU6Ki8qLyoiXSwiY2xpZW50X2lkIjoiVVNFUjMiLCJjaWQiOiJVU0VSMyIsImF6cCI6IlVTRVIzIiwidXNlcl9pZCI6IlVTRVIzIiwiYXBwX2lkIjoidGlsZWNoYXQiLCJpYXQiOjE2MjM3Njc1MjAsImV4cCI6MTkzNDgwNzUyMCwiYXVkIjpbInJhYmJpdG1xIiwiVVNFUjMiXSwia2lkIjoidGlsZWRlc2sta2V5IiwidGlsZWRlc2tfYXBpX3JvbGVzIjoidXNlciJ9.-Cio8ITPCQswv_4KnxJrRbm-5RCXMefuT91wWUNZJmU'
token: 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJqdGkiOiI0NGUzZjdhZC1jNGM1LTQxZmMtOTQzZi0wZjFjZjYwZTBkNDEiLCJzdWIiOiJVU0VSMiIsInNjb3BlIjpbInJhYmJpdG1xLnJlYWQ6Ki8qL2FwcHMudGlsZWNoYXQudXNlcnMuVVNFUjIuKiIsInJhYmJpdG1xLndyaXRlOiovKi9hcHBzLnRpbGVjaGF0LnVzZXJzLlVTRVIyLioiLCJyYWJiaXRtcS53cml0ZToqLyovYXBwcy50aWxlY2hhdC5vdXRnb2luZy51c2Vycy5VU0VSMi4qIiwicmFiYml0bXEuY29uZmlndXJlOiovKi8qIl0sImNsaWVudF9pZCI6IlVTRVIyIiwiY2lkIjoiVVNFUjIiLCJhenAiOiJVU0VSMiIsInVzZXJfaWQiOiJVU0VSMiIsImFwcF9pZCI6InRpbGVjaGF0IiwiaWF0IjoxNjQ0Njc1NzcxLCJleHAiOjE5NTU3MTU3NzEsImF1ZCI6WyJyYWJiaXRtcSIsIlVTRVIyIl0sImtpZCI6InRpbGVkZXNrLWtleSIsInRpbGVkZXNrX2FwaV9yb2xlcyI6InVzZXIifQ.NQsVvyrwGaCz9W6vS1-QSPRxBL1b2mPz1ntLtEFJm_A'
};

let chatClient1 = new Chat21Client(
Expand Down Expand Up @@ -131,7 +123,6 @@ describe("Performance Test", function() {
);
});
});

});

after(function(done) {
Expand All @@ -144,7 +135,6 @@ describe("Performance Test", function() {

it("Benchmark for direct messages", function(done) {
this.timeout(1000 * 70);

async function benchmark() {
console.log("\n\n*********************************************");
console.log("********* Direct messages benchmark *********");
Expand All @@ -168,6 +158,7 @@ describe("Performance Test", function() {
for (let c = 0; c < config.CONCURRENCY; c++) {
let recipient_id = user2.userid;
let recipient_fullname = user2.fullname;
// console.log("sending...",i,c);
sendMessage(i, c, recipient_id, recipient_fullname, async function(latency, iteration, concurrent_iteration) {
// console.log("Direct - latency:", latency)
if (iteration == total_iterations - 1 && concurrent_iteration == config.CONCURRENCY - 1) {
Expand All @@ -183,6 +174,7 @@ describe("Performance Test", function() {
console.log("End 'Direct' benchmark iterations.");

function endCallback(latency) {
console.log("\n\n********* Direct - Benchmark results *********");
console.log("Direct - Final latency:", latency.meanLatencyMs);
let test_duration = Math.round(current / 1000)
console.log("Direct - Test duration:", test_duration + " seconds" + " (" + current + ") ms");
Expand Down Expand Up @@ -241,6 +233,7 @@ describe("Performance Test", function() {
}

function endCallback(latency) {
console.log("\n\n********* Group - Benchmark results *********");
console.log("Group - Final latency:", latency.meanLatencyMs);
console.log("Group - Expected max average latency:", config.EXPECTED_AVG_GROUP_MESSAGE_DELAY);
let test_duration = Math.round(current / 1000)
Expand Down Expand Up @@ -286,7 +279,7 @@ function sendMessage(iteration, concurrent_iteration, recipient_id, recipient_fu
total_delay += delay;
// console.log("total:", total_delay)
let mean = total_delay / total_messages
// console.log("message N:", messages, "currentTimeMs:", current, "meanMs:", Math.floor(mean));
// console.log("total_messages N:", total_messages, "currentTimeMs:", current, "meanMs:", Math.floor(mean));
let latency_info = {
totalMessages: total_messages,
latencyMs: delay,
Expand All @@ -311,7 +304,12 @@ function sendMessage(iteration, concurrent_iteration, recipient_id, recipient_fu
null,
recipient_id.startsWith("group-") ? 'group' : 'direct', //user2.userid.startsWith("group-") ? 'group' : 'direct',
(err, msg) => {
// console.log("Message sent", msg.text);
if (err) {
console.error("Error send:", err);
}
// else {
// console.log("Message sent:", msg.text);
// }
}
);
}
Expand Down
23 changes: 20 additions & 3 deletions chatdb/index.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -118,16 +118,33 @@ class ChatDB {
});
}

getConversation(timelineOf, conversWith, callback) {
this.db.collection(this.conversations_collection).findOne( { timelineOf: timelineOf, conversWith: conversWith }, function(err, doc) {
// DEPRECATED? //
// getConversation(timelineOf, conversWith, callback) {
// this.db.collection(this.conversations_collection).findOne( { timelineOf: timelineOf, conversWith: conversWith }, function(err, doc) {
// if (err) {
// if (callback) {
// callback(err, null)
// }
// }
// else {
// if (callback) {
// callback(null, doc)
// }
// }
// });
// }

conversationDetail(appid, timelineOf, conversWith, archived, callback) {
logger.debug("DB. app: "+ appid+ " user: " + timelineOf + " conversWith: "+ conversWith);
this.db.collection(this.conversations_collection).find( { timelineOf: timelineOf, app_id: appid, conversWith: conversWith, archived: archived } ).limit(1).toArray(function(err, docs) {
if (err) {
if (callback) {
callback(err, null)
}
}
else {
if (callback) {
callback(null, doc)
callback(null, docs)
}
}
});
Expand Down
44 changes: 29 additions & 15 deletions observer.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -842,28 +842,42 @@ function process_archive(topic, payload, callback) {
callback(false)
return
}
const conversation_deleted_topic = 'apps.tilechat.users.' + user_id + '.conversations.' + convers_with + '.clientdeleted'
logger.debug(">>> NOW PUBLISHING... CONVERSATION ARCHIVED (DELETED) TOPIC " + conversation_deleted_topic)
const payload = JSON.stringify(conversation_archive_patch)
publish(exchange, conversation_deleted_topic, Buffer.from(payload), function(err) {
logger.debug(">>> PUBLISHED!!!! CONVERSATION ON TOPIC: " + conversation_deleted_topic + " ARCHIVED (DELETED). Payload: " + payload + " buffered:" + Buffer.from(payload))
chatdb.conversationDetail(app_id, me, convers_with, true, (err, convs) => {
if (err) {
logger.error("error",err);
callback(false)
logger.error("Error getting conversationDetail()", err);
callback(true);
}
else if (convs && convs.length < 1) {
logger.error("Error getting conversationDetail(): convs[].length < 1");
callback(true);
}
else {
// now publish new archived conversation added
const archived_conversation_added_topic = 'apps.tilechat.users.' + user_id + '.archived_conversations.' + convers_with + '.clientadded'
logger.debug(">>> NOW PUBLISHING... CONVERSATION ARCHIVED (ADDED) TOPIC: "+ archived_conversation_added_topic)
// const success_payload = JSON.stringify({"success": true})
publish(exchange, archived_conversation_added_topic, Buffer.from(payload), function(err) {
logger.debug(">>> PUBLISHED!!!! ARCHIVED (DELETED) CONVERSATION ON TOPIC: " + conversation_deleted_topic)
const conversation_archived = convs[0];
logger.debug("got archived conversation detail:", conversation_archived);
const conversation_deleted_topic = 'apps.tilechat.users.' + user_id + '.conversations.' + convers_with + '.clientdeleted'
logger.debug(">>> NOW PUBLISHING... CONVERSATION ARCHIVED (DELETED) TOPIC " + conversation_deleted_topic)
const payload = JSON.stringify(conversation_archived);
publish(exchange, conversation_deleted_topic, Buffer.from(payload), function(err) {
logger.debug(">>> PUBLISHED!!!! CONVERSATION ON TOPIC: " + conversation_deleted_topic + " ARCHIVED (DELETED). Payload: " + payload + " buffered:" + Buffer.from(payload))
if (err) {
logger.error("error",err);
callback(false)
callback(false);
}
else {
callback(true)
// now publish new archived conversation added
const archived_conversation_added_topic = 'apps.tilechat.users.' + user_id + '.archived_conversations.' + convers_with + '.clientadded'
logger.debug(">>> NOW PUBLISHING... CONVERSATION ARCHIVED (ADDED) TOPIC: " + archived_conversation_added_topic)
// const success_payload = JSON.stringify({"success": true})
publish(exchange, archived_conversation_added_topic, Buffer.from(payload), function(err) {
logger.debug(">>> PUBLISHED!!!! ARCHIVED (DELETED) CONVERSATION ON TOPIC: " + conversation_deleted_topic)
if (err) {
logger.error("error",err);
callback(false);
}
else {
callback(true);
}
});
}
});
}
Expand Down
4 changes: 2 additions & 2 deletions package.json
View file
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
{
"name": "@chat21/chat21-server",
"version": "0.2.10",
"description": "Chat21 AMQP messages dispatcher",
"version": "0.2.11",
"description": "Chat21 AMQP/MQTT messages dispatcher",
"dependencies": {
"aedes": "^0.40.1",
"amqplib": "^0.7.1",
Expand Down
5 changes: 3 additions & 2 deletions test/test.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -1583,11 +1583,11 @@ REUSE SHARED CHAT CLIENTS', function(done) {
topic.conversWith === user1.userid) {
// chatClient2.removeOnConversationAddedHandler(handler);
chatClient2.conversationDetail(topic.conversWith, (err, conv) => {
console.log("conv detail:", conv);
logger.log("conv detail:", conv);
assert(err == null);
assert(conv != null);
chatClient2.archivedConversationDetail(topic.conversWith, (err, conv) => {
console.log("archived conv detail:", conv);
logger.log("archived conv detail:", conv);
assert(err == null);
assert(conv == null);
chatClient2.archiveConversation(topic.conversWith, (err) => {
Expand All @@ -1600,6 +1600,7 @@ REUSE SHARED CHAT CLIENTS', function(done) {
let onArchivedConversationAddedHandler = chatClient2.onArchivedConversationAdded((archived_conv, topic) => {
console.log("conv was archived:", archived_conv, topic);
assert(archived_conv != null);
assert(archived_conv.last_message_text != null); // checking archived-conversation is metadata-full
chatClient2.conversationDetail(topic.conversWith, (err, conv) => {
console.log("conv detail:", conv);
assert(err == null);
Expand Down

0 comments on commit cdfa0eb

Please sign in to comment.