Merge branch 'main' into sync-bli

Author: tysonrm

forkrun.c

@@ -1,31 +1,30 @@
 #include <stdio.h>
 #include <stdlib.h>
-#include <unistd.h>
 #include <sys/types.h>
+#include <unistd.h>
 
 #define BUFSIZE 256
 
-char* envvar_cmd = "FORKRUN_CMD";
-char* envvar_arg = "FORKRUN_ARG";
+char *envvar_cmd = "FORKRUN_CMD";
+char *envvar_arg = "FORKRUN_ARG";
 char cmd[BUFSIZE];
 char arg[BUFSIZE];
 
-int main (void) {
-    pid_t pid;    
-    
-    if ((pid = fork()) == 0) {
-        snprintf(cmd, BUFSIZE, "%s", getenv(envvar_cmd));
-        snprintf(arg, BUFSIZE, "%s", getenv(envvar_arg));
-        char* const argv[] = { cmd, arg, NULL };
+int main(void) {
+  pid_t pid;
+  if ((pid = fork()) == 0) {
+    snprintf(cmd, BUFSIZE, "%s", getenv(envvar_cmd));
+    snprintf(arg, BUFSIZE, "%s", getenv(envvar_arg));
+    char *const argv[] = {cmd, arg, NULL};
 
-        if (execv(cmd, argv) < 0) {
-            perror("execv error");
-        }
-    } else if (pid < 0) {
-        perror("fork error");
-    } else {
-        return EXIT_SUCCESS;
+    if (execv(cmd, argv) < 0) {
+      perror("execv error");
     }
+  } else if (pid < 0) {
+    perror("fork error");
+  } else {
+    return EXIT_SUCCESS;
+  }
 
-    return EXIT_FAILURE;
+  return EXIT_FAILURE;
 }

package.json

@@ -23,15 +23,15 @@
     "core-js": "^3"
   },
   "dependencies": {
-    "@assemblyscript/loader": "0.21.1",
+    "@assemblyscript/loader": "0.25.2",
     "@octokit/core": "4.0.5",
-    "@octokit/rest": "19.0.4",
+    "@octokit/rest": "19.0.5",
     "acme-client": "^5.0.0",
     "as-bind": "0.8.2",
-    "assemblyscript": "0.21.1",
+    "assemblyscript": "0.25.2",
     "cors": "^2.8.5",
     "dotenv": "16.0.1",
-    "express": "4.18.1",
+    "express": "4.18.2",
     "express-fileupload": "^1.4.0",
     "express-jwt": "7.7.5",
     "import-fresh": "3.3.0",
@@ -59,7 +59,7 @@
     "@babel/core": "7.18.10",
     "@babel/node": "7.18.10",
     "@babel/preset-env": "7.18.10",
-    "@types/node": "16.11.52",
+    "@types/node": "16.18.11",
     "babel-loader": "8.2.5",
     "chalk": "5.0.1",
     "cross-var": "1.1.0",

src/adapters/datasources/datasource-mongodb.js

@@ -21,11 +21,12 @@ const processQuery = qpm({
 const url = process.env.MONGODB_URL || 'mongodb://localhost:27017'
 
 const dsOptions = configRoot.adapters.datasources.DataSourceMongoDb.options || {
-  runOffline: false,
-  numConns: 2,
+  runOffline: true,
+  numConns: 2
 }
 
 const cacheSize = configRoot.adapters.cacheSize || 3000
+let connPool
 
 /**
  * @type {Map<string,MongoClient>}
@@ -43,7 +44,7 @@ const mongoOpts = {
  * even when the database is offline.
  */
 export class DataSourceMongoDb extends DataSource {
-  constructor(map, name, namespace, options = {}) {
+  constructor (map, name, namespace, options = {}) {
     super(map, name, namespace, options)
     this.cacheSize = cacheSize
     this.mongoOpts = mongoOpts
@@ -52,47 +53,56 @@ export class DataSourceMongoDb extends DataSource {
     this.url = url
   }
 
-  connect(client) {
-    return async function () {
-      let timeout = false
-      const timerId = setTimeout(() => {
-        timeout = true
-      }, 500)
-      await client.connect()
-      clearTimeout(timerId)
-      if (timeout) throw new Error('mongo conn timeout')
-    }
-  }
-
-  async connectionPool() {
+  /**
+   *
+   * @returns {Promise<import('mongodb').Db>}
+   */
+  async connectionPool () {
     return new Promise((resolve, reject) => {
       if (this.db) return resolve(this.db)
       MongoClient.connect(
         this.url,
         {
           ...this.mongoOpts,
           poolSize: dsOptions.numConns || 2,
+          connectTimeoutMS: 500
         },
-        (err, database) => {
+        (err, db) => {
           if (err) return reject(err)
-          resolve((this.db = database.db(this.namespace)))
+          this.db = db(this.namespace)
+          resolve(this.db)
         }
       )
     })
   }
 
-  async connection() {
+  connect (client) {
+    return async function () {
+      let timeout = false
+      const timerId = setTimeout(() => {
+        timeout = true
+      }, 500)
+      await client.connect()
+      clearTimeout(timerId)
+      if (timeout) throw new Error('mongo conn timeout')
+    }
+  }
+
+  async connection () {
     try {
       while (connections.length < (dsOptions.numConns || 1)) {
-        const client = new MongoClient(this.url, this.mongoOpts)
+        const client = new MongoClient(this.url, {
+          ...this.mongoOpts,
+          connectTimeoutMS: 500
+        })
         const thresholds = {
           default: {
             errorRate: 1,
             callVolume: 1,
             intervalMs: 10000,
             testDelay: 300000,
-            //fallbackFn: () => client.emit('connectionClosed')
-          },
+            fallbackFn: () => console.log('circuit open')
+          }
         }
         const breaker = CircuitBreaker(
           'mongodb.connect',
@@ -145,7 +155,7 @@ export class DataSourceMongoDb extends DataSource {
       .createIndexes(indexOperations)
   }
 
-  async find(id) {
+  async find (id) {
     try {
       return (await this.collection()).findOne({ _id: id })
     } catch (error) {
@@ -163,7 +173,7 @@ export class DataSourceMongoDb extends DataSource {
    * @param {*} id
    * @param {*} data
    */
-  async save(id, data) {
+  async save (id, data) {
     try {
       await (
         await this.collection()
@@ -186,19 +196,19 @@ export class DataSourceMongoDb extends DataSource {
    * @param {number} highWaterMark num of docs per batch write
    * @returns
    */
-  createWriteStream(filter = {}, highWaterMark = HIGHWATERMARK) {
+  createWriteStream (filter = {}, highWaterMark = HIGHWATERMARK) {
     try {
       let objects = []
       const ctx = this
 
-      async function upsert() {
+      async function upsert () {
         const operations = objects.map(obj => {
           return {
             replaceOne: {
               filter: { ...filter, _id: obj.id },
               replacement: { ...obj, _id: obj.id },
-              upsert: true,
-            },
+              upsert: true
+            }
           }
         })
 
@@ -217,17 +227,17 @@ export class DataSourceMongoDb extends DataSource {
       const writable = new Writable({
         objectMode: true,
 
-        async write(chunk, _encoding, next) {
+        async write (chunk, _encoding, next) {
           objects.push(chunk)
           // if true time to flush buffer and write to db
           if (objects.length >= highWaterMark) await upsert()
           next()
         },
 
-        end(chunk, _, done) {
+        end (chunk, _, done) {
           objects.push(chunk)
           done()
-        },
+        }
       })
 
       writable.on('finish', async () => await upsert())
@@ -247,7 +257,7 @@ export class DataSourceMongoDb extends DataSource {
    *
    * @returns {Promise<import('mongodb').AbstractCursor>}
    */
-  async mongoFind({ filter, sort, limit, aggregate, skip } = {}) {
+  async mongoFind ({ filter, sort, limit, aggregate, skip } = {}) {
     console.log({ fn: this.mongoFind.name, filter })
     let cursor = aggregate
       ? (await this.collection()).aggregate(aggregate)
@@ -287,7 +297,6 @@ export class DataSourceMongoDb extends DataSource {
    */
   streamList({ writable, serialize, transform, options }) {
     try {
-      let pipeArgs = []
 
       const serializer = new Transform({
         writableObjectMode: true,
@@ -340,15 +349,12 @@ export class DataSourceMongoDb extends DataSource {
       })
 
       return new Promise(async (resolve, reject) => {
+        const pipeArgs = []
         const readable = (await this.mongoFind(options)).stream()
 
-        readable.on('error', reject)
-        readable.on('end', resolve)
-
         // optionally transform db stream then pipe to output
         pipeArgs.push(readable)
-        if (transform) pipeArgs.push(transform)
-        if (serialize) pipeArgs.push(serializer)
+        
         if (options.page) {
           const count = ~~(await this.mongoCount(options.filter))
           pipeArgs.push(paginate)
@@ -360,10 +366,9 @@ export class DataSourceMongoDb extends DataSource {
             () => console.log('paginated query', options)
           )
         }
-        pipeArgs.push(writable)
-
-        // perform pipe operations
-        pipeArgs.reduce((prevVal, currVal) => prevVal.pipe(currVal))
+        
+        return pipeArgs
+   
       })
     } catch (error) {
       console.error({
@@ -382,7 +387,7 @@ export class DataSourceMongoDb extends DataSource {
    * @override
    * @param {import('../../domain/datasource').listOptions} param
    */
-  async list(param) {
+  async list (param) {
     try {
       const options = this.processOptions(param)
       if (0 < ~~query.__page) {
@@ -401,7 +406,7 @@ export class DataSourceMongoDb extends DataSource {
       }
       console.log({ options })
 
-      if (writable && !query.__json)
+      if (streamRequested && !query.__json)
         this.streamList({ writable, serialize, transform, options })
       else {
         const data = (await this.mongoFind(options)).toArray()
@@ -424,19 +429,19 @@ export class DataSourceMongoDb extends DataSource {
    *
    * @override
    */
-  async count() {
+  async count () {
     return {
       total: await this.countDb(),
       cached: this.getCacheSize(),
-      bytes: this.getCacheSizeBytes(),
+      bytes: this.getCacheSizeBytes()
     }
   }
 
   /**
    * @override
    * @returns
    */
-  async countDb() {
+  async countDb () {
     return (await this.collection()).countDocuments()
   }
 
@@ -447,7 +452,7 @@ export class DataSourceMongoDb extends DataSource {
    * @override
    * @param {*} id
    */
-  async delete(id) {
+  async delete (id) {
     try {
       await (await this.collection()).deleteOne({ _id: id })
     } catch (error) {