puemos · May 6, 2018 07:53 · May 6, 2018 · May 6, 2018 · May 6, 2018 · May 6, 2018
diff --git a/create-concurrent-task-queue.js b/create-concurrent-task-queue.js
@@ -9,28 +9,29 @@ import { all, call, fork, put, take } from "redux-saga/effects";
  */
 function* createConcurrentTaskQueue(handler, workersCount = 1) {
 
-  // create a channel to queue incoming action
+  // a channel to queue incoming action
   const queueChannel = yield call(channel, buffers.expanding());
 
-  // watch for actions and delegate them as tasks to the workers
   function* watcher() {
-    // create a channel to queue incoming tasks
+    // a channel to queue incoming tasks
     const workersChannel = yield call(channel, buffers.expanding());
 
     // create n worker 'threads'
     yield all(Array(workersCount).fill(fork(worker, workersChannel)));
 
-    // wait for a task and once it arrive assign it to one of the workers
+    // wait for a tasks
     while (true) {
+      // incoming task
       const { payload } = yield take(queueChannel);
+      // assign the task to one of the workers
       yield put(workersChannel, payload);
     }
   }
 
-  // worker that waits for a task
+  // a single worker
   function* worker(chan) {
     while (true) {
-      // take an task
+      // incoming task
       const payload = yield take(chan);
       // handle it with the given handler arg
       yield handler(payload);

diff --git a/create-concurrent-task-queue.js b/create-concurrent-task-queue.js
@@ -39,7 +39,7 @@ function* createConcurrentTaskQueue(handler, workersCount = 1) {
 
   return {
     watcher,
-    addTaskChannel,
+    queueChannel,
   };
 }
 

diff --git a/create-concurrent-task-queue.js b/create-concurrent-task-queue.js
@@ -37,14 +37,9 @@ function* createConcurrentTaskQueue(handler, workersCount = 1) {
     }
   }
 
-  // add a task to the queue
-  function* queue(action){
-    yield put(addTaskChannel, action)
-  }
-
   return {
     watcher,
-    queue,
+    addTaskChannel,
   };
 }
 

diff --git a/create-concurrent-task-queue.js b/create-concurrent-task-queue.js
@@ -20,20 +20,24 @@ function* createConcurrentTaskQueue(handler, workersCount = 1) {
     // create n worker 'threads'
     yield all(Array(workersCount).fill(fork(worker, workersChannel)));
 
+    // wait for a task and once it arrive assign it to one of the workers
     while (true) {
       const { payload } = yield take(queueChannel);
       yield put(workersChannel, payload);
     }
   }
 
-  // worker that wait for a task
+  // worker that waits for a task
   function* worker(chan) {
     while (true) {
+      // take an task
       const payload = yield take(chan);
+      // handle it with the given handler arg
       yield handler(payload);
     }
   }
 
+  // add a task to the queue
   function* queue(action){
     yield put(addTaskChannel, action)
   }

diff --git a/createConcurrentTaskQueue.js → create-concurrent-task-queue.js b/createConcurrentTaskQueue.js → create-concurrent-task-queue.js
diff --git a/msgs.js b/msgs.js
@@ -1,11 +0,0 @@
-function* handler(payload){
-  ...
-}
-
-function* bulkMsgs(action){
-  const { msgs } = action.payload
-  const concurrentTaskQueue = yield createConcurrentTaskQueue(handler, 3)
-  for (msg in msgs){
-    concurrentTaskQueue.queue({ type: 'SEND', payload: msg })
-  }
-}

diff --git a/msgs.js b/msgs.js
@@ -0,0 +1,11 @@
+function* handler(payload){
+  ...
+}
+
+function* bulkMsgs(action){
+  const { msgs } = action.payload
+  const concurrentTaskQueue = yield createConcurrentTaskQueue(handler, 3)
+  for (msg in msgs){
+    concurrentTaskQueue.queue({ type: 'SEND', payload: msg })
+  }
+}
diff --git a/createConcurrentTaskQueue.js b/createConcurrentTaskQueue.js
@@ -12,6 +12,7 @@ function* createConcurrentTaskQueue(handler, workersCount = 1) {
   // create a channel to queue incoming action
   const queueChannel = yield call(channel, buffers.expanding());
 
+  // watch for actions and delegate them as tasks to the workers
   function* watcher() {
     // create a channel to queue incoming tasks
     const workersChannel = yield call(channel, buffers.expanding());
@@ -25,6 +26,7 @@ function* createConcurrentTaskQueue(handler, workersCount = 1) {
     }
   }
 
+  // worker that wait for a task
   function* worker(chan) {
     while (true) {
       const payload = yield take(chan);

diff --git a/createConcurrentTaskQueue.js b/createConcurrentTaskQueue.js
@@ -5,9 +5,9 @@ import { all, call, fork, put, take } from "redux-saga/effects";
  * creates a queue
  *
  * @param {GeneratorFunction} [handler] request handler
- * @param {number} [concurrent=1] number of workers
+ * @param {number} [workersCount=1] number of workers
  */
-function* createConcurrentTaskQueue(handler, concurrent = 1) {
+function* createConcurrentTaskQueue(handler, workersCount = 1) {
 
   // create a channel to queue incoming action
   const queueChannel = yield call(channel, buffers.expanding());
@@ -17,7 +17,7 @@ function* createConcurrentTaskQueue(handler, concurrent = 1) {
     const workersChannel = yield call(channel, buffers.expanding());
 
     // create n worker 'threads'
-    yield all(Array(concurrent).fill(fork(worker, workersChannel)));
+    yield all(Array(workersCount).fill(fork(worker, workersChannel)));
 
     while (true) {
       const { payload } = yield take(queueChannel);

diff --git a/queue1.js → createConcurrentTaskQueue.js b/queue1.js → createConcurrentTaskQueue.js
diff --git a/queue1.js b/queue1.js
@@ -7,21 +7,21 @@ import { all, call, fork, put, take } from "redux-saga/effects";
  * @param {GeneratorFunction} [handler] request handler
  * @param {number} [concurrent=1] number of workers
  */
-function* ConcurrentTaskQueue(handler, concurrent = 1) {
+function* createConcurrentTaskQueue(handler, concurrent = 1) {
+
+  // create a channel to queue incoming action
   const queueChannel = yield call(channel, buffers.expanding());
+
   function* watcher() {
-    try {
-      // create a channel to queue incoming requests
-      const workersChannel = yield call(channel, buffers.expanding());
+    // create a channel to queue incoming tasks
+    const workersChannel = yield call(channel, buffers.expanding());
 
-      // create n worker 'threads'
-      yield all(Array(concurrent).fill(fork(worker, workersChannel)));
+    // create n worker 'threads'
+    yield all(Array(concurrent).fill(fork(worker, workersChannel)));
 
-      while (true) {
-        const { payload } = yield take(queueChannel);
-        yield put(workersChannel, payload);
-      }
-    } finally {
+    while (true) {
+      const { payload } = yield take(queueChannel);
+      yield put(workersChannel, payload);
     }
   }
 
@@ -42,4 +42,4 @@ function* ConcurrentTaskQueue(handler, concurrent = 1) {
   };
 }
 
-export default ConcurrentTaskQueue;
+export default createConcurrentTaskQueue;
diff --git a/queue1.js b/queue1.js
@@ -42,4 +42,4 @@ function* ConcurrentTaskQueue(handler, concurrent = 1) {
   };
 }
 
-export default createQueue;
+export default ConcurrentTaskQueue;
diff --git a/queue1.js b/queue1.js
@@ -8,33 +8,37 @@ import { all, call, fork, put, take } from "redux-saga/effects";
  * @param {number} [concurrent=1] number of workers
  */
 function* ConcurrentTaskQueue(handler, concurrent = 1) {
-  const addTaskChannel = yield call(channel, buffers.expanding());
-  function* watchRequests() {
+  const queueChannel = yield call(channel, buffers.expanding());
+  function* watcher() {
     try {
       // create a channel to queue incoming requests
-      const runChannel = yield call(channel, buffers.expanding());
+      const workersChannel = yield call(channel, buffers.expanding());
 
       // create n worker 'threads'
-      yield all(Array(concurrent).fill(fork(handleRequest, runChannel)));
+      yield all(Array(concurrent).fill(fork(worker, workersChannel)));
 
       while (true) {
-        const { payload } = yield take(addTaskChannel);
-        yield put(runChannel, payload);
+        const { payload } = yield take(queueChannel);
+        yield put(workersChannel, payload);
       }
     } finally {
     }
   }
 
-  function* handleRequest(chan) {
+  function* worker(chan) {
     while (true) {
       const payload = yield take(chan);
       yield handler(payload);
     }
   }
-
+
+  function* queue(action){
+    yield put(addTaskChannel, action)
+  }
+
   return {
-    watcher: watchRequests,
-    addTaskChannel
+    watcher,
+    queue,
   };
 }
 

diff --git a/queue1.js b/queue1.js
@@ -7,7 +7,7 @@ import { all, call, fork, put, take } from "redux-saga/effects";
  * @param {GeneratorFunction} [handler] request handler
  * @param {number} [concurrent=1] number of workers
  */
-function* createQueue(handler, concurrent = 1) {
+function* ConcurrentTaskQueue(handler, concurrent = 1) {
   const addTaskChannel = yield call(channel, buffers.expanding());
   function* watchRequests() {
     try {

diff --git a/queue1.js b/queue1.js
@@ -4,10 +4,10 @@ import { all, call, fork, put, take } from "redux-saga/effects";
 /**
  * creates a queue
  *
- * @param {GeneratorFunction} [handle=() => {}] request handler
+ * @param {GeneratorFunction} [handler] request handler
  * @param {number} [concurrent=1] number of workers
  */
-function* createQueue(handle = () => {}, concurrent = 1) {
+function* createQueue(handler, concurrent = 1) {
   const addTaskChannel = yield call(channel, buffers.expanding());
   function* watchRequests() {
     try {
@@ -28,7 +28,7 @@ function* createQueue(handle = () => {}, concurrent = 1) {
   function* handleRequest(chan) {
     while (true) {
       const payload = yield take(chan);
-      yield handle(payload);
+      yield handler(payload);
     }
   }
 

diff --git a/queue1.js b/queue1.js
@@ -0,0 +1,41 @@
+import { buffers, channel } from "redux-saga";
+import { all, call, fork, put, take } from "redux-saga/effects";
+
+/**
+ * creates a queue
+ *
+ * @param {GeneratorFunction} [handle=() => {}] request handler
+ * @param {number} [concurrent=1] number of workers
+ */
+function* createQueue(handle = () => {}, concurrent = 1) {
+  const addTaskChannel = yield call(channel, buffers.expanding());
+  function* watchRequests() {
+    try {
+      // create a channel to queue incoming requests
+      const runChannel = yield call(channel, buffers.expanding());
+
+      // create n worker 'threads'
+      yield all(Array(concurrent).fill(fork(handleRequest, runChannel)));
+
+      while (true) {
+        const { payload } = yield take(addTaskChannel);
+        yield put(runChannel, payload);
+      }
+    } finally {
+    }
+  }
+
+  function* handleRequest(chan) {
+    while (true) {
+      const payload = yield take(chan);
+      yield handle(payload);
+    }
+  }
+
+  return {
+    watcher: watchRequests,
+    addTaskChannel
+  };
+}
+
+export default createQueue;
No results found