Merge pull request #11209 from opencrvs/stream-batch-sizing-b

Stream batch sizing

Author: rikukissa

packages/events/src/index.ts

@@ -26,8 +26,8 @@ const appModulePath = require('app-module-path')
 appModulePath.addPath(path.join(__dirname, '../'))
 
 export async function main() {
-  await ensureConnection()
   try {
+    await ensureConnection()
     const anonymousToken = await getAnonymousToken()
     const configurations = await getInMemoryEventConfigurations(
       `Bearer ${anonymousToken}`

packages/events/src/service/events/reindex.ts

@@ -9,6 +9,7 @@
  * Copyright (C) The OpenCRVS Authors located at https://github.com/opencrvs/opencrvs-core/blob/master/AUTHORS.
  */
 import { Readable, Transform, PassThrough } from 'node:stream'
+import { Agent } from 'node:http'
 import fetch from 'node-fetch'
 import { JsonStreamStringify } from 'json-stream-stringify'
 import { EventDocument } from '@opencrvs/commons/events'
@@ -98,7 +99,10 @@ export async function reindex(token: TokenWithBearer) {
       },
       // Converts object stream to JSON string stream so that it can
       // be sent to the country config reindex endpoint
-      body: new JsonStreamStringify(eventDocumentStreamForCountryConfig)
+      body: new JsonStreamStringify(eventDocumentStreamForCountryConfig),
+      // Ensure HTTP socket of previous GET /events request is not reused
+      // to avoid connections being closed preemptively by Node.js
+      agent: new Agent({ keepAlive: false })
     }
   ).then((response) => {
     if (!response.ok && response.status === 404) {