Merge pull request #264 from DavidVujic/libuv-assert

Example code: use one client, use only when connected

Author: DavidVujic

examples/addlistener.js

@@ -1,3 +1,4 @@
+const { isClientConnected } = require('./wrapper.js');
 const notifier = require('./notifier.js');
 const logger = require('./logger.js');
 
@@ -18,10 +19,13 @@ async function listen(client, path) {
     const watchFunc = watcher.bind(null, client, listen);
 
     try {
+        if (!isClientConnected()) {
+            throw new Error('listen: client is not connected');
+        }
         const children = await client.w_get_children(path, watchFunc);
         emit(client, path, children);
     } catch (error) {
-        logger.error(error);
+        logger.error('listen', error.message);
     }
 }
 

examples/addtask.js

@@ -1,4 +1,4 @@
-const { constants, createClient } = require('./wrapper.js');
+const { constants } = require('./wrapper.js');
 const notifier = require('./notifier.js');
 const { createNode, persistentNode } = require('./createnode.js');
 
@@ -8,15 +8,8 @@ async function createTask(client, data) {
     notifier.emit('addTask', message);
 }
 
-async function addTask(data) {
-    const client = createClient();
-
-    client.on('connect', () => {
-        notifier.emit('connect', `addTask: session established, id=${client.client_id}`);
-
-        createTask(client, data);
-    });
-    client.init({});
+async function addTask(client, data) {
+    createTask(client, data);
 }
 
 module.exports = {

examples/createnode.js

@@ -1,3 +1,5 @@
+const { isClientConnected } = require('./wrapper.js');
+
 const persistentNode = 0;
 
 /**
@@ -9,10 +11,14 @@ const persistentNode = 0;
  */
 async function createNode(client, path, flags, data = '') {
     try {
+        if (!isClientConnected()) {
+            throw new Error('createNode: client is not connected');
+        }
+
         const createdPath = await client.create(path, data, flags);
         return `(created: ${createdPath})`;
     } catch (error) {
-        return `${path} already exists`;
+        return `${path} Error: ${error.message}`;
     }
 }
 

examples/createworker.js

@@ -1,31 +1,27 @@
-const { constants, createClient } = require('./wrapper.js');
+const { constants, isClientConnected } = require('./wrapper.js');
 const notifier = require('./notifier.js');
 const logger = require('./logger.js');
 
 function emit(client, path) {
     logger.log(`(${path}) client id: ${client.client_id}`);
-    notifier.emit('createWorker', client);
+    notifier.emit('createWorker');
 }
 
 async function createWorkerPath(client, path) {
     try {
+        if (!isClientConnected()) {
+            throw new Error('createWorkerPath: client is not connected');
+        }
         // eslint-disable-next-line no-bitwise
         const createdPath = await client.create(path, '', constants.ZOO_EPHEMERAL | constants.ZOO_SEQUENCE);
         emit(client, createdPath);
     } catch (error) {
-        logger.error(error);
+        logger.error('createWorkerPath', error.message);
     }
 }
 
-async function createWorker() {
-    const client = createClient();
-
-    client.on('connect', () => {
-        notifier.emit('connect', `createWorker: session established, id=${client.client_id}`);
-        createWorkerPath(client, '/workers/worker-');
-    });
-
-    client.init({});
+async function createWorker(client) {
+    createWorkerPath(client, '/workers/worker-');
 }
 
 module.exports = {

examples/electleader.js

@@ -1,10 +1,10 @@
-const { constants, createClient } = require('./wrapper.js');
+const { constants, isClientConnected } = require('./wrapper.js');
 const notifier = require('./notifier.js');
 const logger = require('./logger.js');
 
 function emit(client, path) {
-    logger.log(`(${path}) ${client.client_id}`);
-    notifier.emit('leader', client);
+    logger.log(`Elect leader: (${path}) ${client.client_id}`);
+    notifier.emit('leader');
 }
 
 function onData(client, path, rc, error, stat, data) {
@@ -27,33 +27,33 @@ async function checkMaster(client, path, retryFunc) {
     const watchFunc = watcher.bind(null, client, path, checkMaster, retryFunc);
 
     try {
+        if (!isClientConnected()) {
+            throw new Error('is not connected');
+        }
         const res = await client.w_get(path, watchFunc);
         onData(client, path, res.rc, res.error, res.stat, res.data);
     } catch (error) {
-        logger.error(error);
+        logger.error('checkMaster:', error.message);
     }
 }
 
 async function runForLeader(client, path) {
     const clientId = client.client_id;
 
     try {
+        if (!isClientConnected()) {
+            throw new Error('is not connected');
+        }
         await client.create(path, `${clientId}`, constants.ZOO_EPHEMERAL);
         emit(client, path);
     } catch (error) {
+        logger.error('runForLeader:', error.message);
         await checkMaster(client, path, runForLeader);
     }
 }
 
-async function electLeader(path) {
-    const client = createClient();
-
-    client.on('connect', () => {
-        notifier.emit('connect', `electLeader: session established, id=${client.client_id}`);
-        runForLeader(client, path);
-    });
-
-    client.init({});
+async function electLeader(client, path) {
+    runForLeader(client, path);
 }
 
 module.exports = {

examples/index.js

@@ -1,3 +1,4 @@
+const { getClient } = require('./wrapper.js');
 const { createNodes } = require('./setup.js');
 const { electLeader } = require('./electleader.js');
 const { createWorker } = require('./createworker.js');
@@ -10,7 +11,6 @@ const notifier = require('./notifier.js');
 notifier.on('connect', (message) => logger.log('connect', message));
 notifier.on('createNode', (message) => logger.log('createNode', message));
 notifier.on('addTask', (message) => logger.log('addTask', message));
-notifier.on('close', (message) => logger.log('close', message));
 
 notifier.on('onChildren', (children) => {
     children.forEach((child) => {
@@ -19,21 +19,25 @@ notifier.on('onChildren', (children) => {
 });
 
 async function init() {
-    await createNodes(['/workers', '/assign', '/tasks', '/status']);
+    const client = getClient();
 
-    notifier.on('leader', async (master) => {
-        await listen(master, '/workers');
-        await listen(master, '/assign');
+    client.on('connect', async () => {
+        await createNodes(client, ['/workers', '/assign', '/tasks', '/status']);
 
-        notifier.on('createWorker', async (worker) => {
-            await listen(worker, '/tasks');
+        notifier.on('leader', async () => {
+            await listen(client, '/workers');
+            await listen(client, '/assign');
+
+            notifier.on('createWorker', async () => {
+                await listen(client, '/tasks');
+            });
+
+            await createWorker(client);
+            await addTask(client, 'hello world');
         });
 
-        await createWorker();
-        await addTask('hello world');
+        await electLeader(client, '/master');
     });
-
-    await electLeader('/master');
 }
 
 init().catch(logger.error);

examples/setup.js

@@ -1,4 +1,3 @@
-const { createClient } = require('./wrapper.js');
 const notifier = require('./notifier.js');
 const { createNode, persistentNode } = require('./createnode.js');
 
@@ -14,20 +13,8 @@ async function createAllNodes(client, paths) {
     });
 }
 
-async function createNodes(paths) {
-    const client = createClient();
-
-    client.on('close', () => {
-        notifier.emit('close', `session closed, id=${client.client_id}`);
-    });
-
-    client.on('connect', () => {
-        notifier.emit('connect', `session established, id=${client.client_id}`);
-
-        createAllNodes(client, paths);
-    });
-
-    client.init({});
+async function createNodes(client, paths) {
+    createAllNodes(client, paths);
 }
 
 module.exports = {

examples/wrapper.js

@@ -1,24 +1,65 @@
 const { constants, Promise: ZooKeeper } = require('../lib/index');
+const logger = require('./logger.js');
 
 const host = process.argv[2] || '127.0.0.1:2181';
 
+let client;
+let isConnected = false;
+
 /**
  * @param timeoutMs {number}
  * @returns {ZooKeeper}
  */
 function createClient(timeoutMs = 5000) {
-    const config = {
-        connect: host,
-        timeout: timeoutMs,
-        debug_level: constants.ZOO_LOG_LEVEL_WARN,
-        host_order_deterministic: false,
-    };
-
-    return new ZooKeeper(config);
+    if (!client) {
+        isConnected = false;
+        logger.log('creating a client.');
+
+        const config = {
+            connect: host,
+            timeout: timeoutMs,
+            debug_level: constants.ZOO_LOG_LEVEL_WARN,
+            host_order_deterministic: false,
+        };
+
+        client = new ZooKeeper(config);
+
+        client.on('close', () => {
+            isConnected = false;
+            logger.log('close', `session closed, id=${client.client_id}`);
+            client = null;
+        });
+
+        client.on('connecting', () => {
+            isConnected = false;
+            logger.log('connecting', `session connecting, id=${client.client_id}`);
+        });
+
+        client.on('connect', () => {
+            isConnected = true;
+            logger.log('connect', `session connect, id=${client.client_id}`);
+        });
+
+        setTimeout(() => {
+            client.init({});
+        }, 1000);
+    }
+
+    return client;
+}
+
+/** @returns {ZooKeeper} */
+function getClient() {
+    return createClient();
+}
+
+function isClientConnected() {
+    return isConnected;
 }
 
 module.exports = {
     constants,
-    createClient,
     ZooKeeper,
+    getClient,
+    isClientConnected,
 };