Revert current concurrency implementation (#147)

Too many problems arose, primarily the [problems with the message order](#22). For now, this implementation is reverted. Reviewed-on: https://git.verdigado.com/NB-Public/rocketchat2matrix/pulls/147 Co-authored-by: Henrik Hüttemann <[email protected]> Co-committed-by: Henrik Hüttemann <[email protected]>
verdigado · Jul 25, 2024 · 698062c · 698062c
1 parent dde8005
commit 698062c
Show file tree

Hide file tree

Showing 4 changed files with 4 additions and 25 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -54,7 +54,6 @@
     "dotenv": "^16.4.5",
     "n-readlines": "^1.0.1",
     "node-emoji": "^2.1.3",
-    "p-limit": "^3.0.0",
     "reflect-metadata": "^0.2.2",
     "showdown": "^2.1.0",
     "sqlite3": "^5.1.7",

diff --git a/src/Entities.ts b/src/Entities.ts
@@ -2,7 +2,6 @@ export const enum Entity {
   Users = 'users',
   Rooms = 'rooms',
   Messages = 'messages',
-  ThreadMessages = 'threadmessages',
 }
 
 type EntityConfig = {
@@ -25,8 +24,4 @@ export const entities: {
     filename: 'rocketchat_message.json',
     mappingType: 2,
   },
-  threadmessages: {
-    filename: 'rocketchat_message.json',
-    mappingType: 2,
-  },
 } as const
diff --git a/src/app.ts b/src/app.ts
@@ -3,7 +3,6 @@ dotenv.config()
 import { AxiosError } from 'axios'
 import lineByLine from 'n-readlines'
 import { exit } from 'node:process'
-import pLimit from 'p-limit'
 import 'reflect-metadata'
 import { Entity, entities } from './Entities'
 import { handleDirectChats } from './handlers/directChats'
@@ -25,9 +24,6 @@ log.info('rocketchat2matrix starts.')
 async function loadRcExport(entity: Entity) {
   const rl = new lineByLine(`./inputs/${entities[entity].filename}`)
 
-  const limit = pLimit(parseInt(process.env.CONCURRENCY_LIMIT || '50'))
-  const queue = []
-
   let line: false | Buffer
   while ((line = rl.next())) {
     const item = JSON.parse(line.toString())
@@ -37,27 +33,17 @@ async function loadRcExport(entity: Entity) {
         break
 
       case Entity.Rooms:
-        queue.push(limit(() => handleRoom(item)))
+        await handleRoom(item)
         break
 
       case Entity.Messages:
-        if (!item.tmid) {
-          queue.push(limit(() => handleMessage(item)))
-        }
-        break
-
-      case Entity.ThreadMessages:
-        if (item.tmid) {
-          queue.push(limit(() => handleMessage(item)))
-        }
+        await handleMessage(item)
         break
 
       default:
         throw new Error(`Unhandled Entity: ${entity}`)
     }
   }
-
-  await Promise.all(queue)
 }
 
 async function main() {
@@ -71,8 +57,6 @@ async function main() {
     await loadRcExport(Entity.Rooms)
     log.info('Parsing messages')
     await loadRcExport(Entity.Messages)
-    log.info('Parsing threaded messages')
-    await loadRcExport(Entity.ThreadMessages)
     log.info('Setting direct chats to be displayed as such for each user')
     await handleDirectChats()
     log.info('Setting pinned messages in rooms')