add lots of proof-of-concept work for schedulor and enhance

kafka-images/deploy.sh

@@ -3,17 +3,22 @@ set -e
 
 wsk=$1
 
-pushd $(dirname $0)/enhance
+pushd >/dev/null $(dirname $0)/schedulor
+zip -ru ../schedulor.zip * || true
+popd >/dev/null
+$wsk -i action update --web true schedulor $(dirname $0)/schedulor.zip --kind nodejs:8
+
+pushd >/dev/null $(dirname $0)/enhance
 zip -ru ../enhance.zip * || true
-popd
+popd >/dev/null
 $wsk -i action update enhance $(dirname $0)/enhance.zip --kind nodejs:8
 
-pushd $(dirname $0)/grayscale
+pushd >/dev/null $(dirname $0)/grayscale
 zip -ru ../grayscale.zip * || true
-popd
+popd >/dev/null
 $wsk -i action update grayscale $(dirname $0)/grayscale.zip --kind nodejs:8
 
-pushd $(dirname $0)/zip
+pushd >/dev/null $(dirname $0)/zip
 zip -ru ../zip.zip * || true
-popd
+popd >/dev/null
 $wsk -i action update zip $(dirname $0)/zip.zip --kind nodejs:8

kafka-images/enhance/index.js

@@ -7,7 +7,7 @@ async function writeToKafka(topic, key, json) {
     brokers: ['kafka03-prod02.messagehub.services.eu-gb.bluemix.net:9093', 'kafka02-prod02.messagehub.services.eu-gb.bluemix.net:9093', 'kafka01-prod02.messagehub.services.eu-gb.bluemix.net:9093'],
     ssl: true,
     sasl: {
-      mechanism: 'scram-sha-256',
+      mechanism: 'plain',
       username: 'urGiSyjzBjRRNAHE',
       password: 'rHICukLEqbSb9DIKrYmejTRRkMgDUkZK'
     }
@@ -19,11 +19,10 @@ async function writeToKafka(topic, key, json) {
     topic: topic,
     messages: [{
       key: key,
-      value: json]
+      value: json
+    }]
   })
   await producer.disconnect()
-
-  return {};
 }
 
 async function processImage(filename, data) {
@@ -35,8 +34,7 @@ async function processImage(filename, data) {
   .toBuffer((err, buffer) => {
     if (err)
       return;
-      writeToKafka("topic3", filename + ".enhanced.jpg", '{"data": "' + buffer.toString("base64") + '"}');
-      resolve({});
+    writeToKafka("topic3", filename + ".enhanced.jpg", '{"data": "' + buffer.toString("base64") + '"}');
   })
 }
 
@@ -44,3 +42,5 @@ async function processImage(filename, data) {
 function main({messages}) {
   messages.forEach(msg => processImage(msg.key, msg.value.data))
 }
+
+exports.main = main

kafka-images/grayscale/index.js

@@ -16,10 +16,11 @@ async function writeToKafka(topic, key, json) {
   const producer = kafka.producer()
   await producer.connect()
   await producer.send({
-    topic: 'mra5q18k-' + topic,
+    topic: topic,
     messages: [{
       key: key,
-      value: json]
+      value: json
+    }]
   })
   await producer.disconnect()
 

kafka-images/schedulor/index.js

+// const openwhisk = require('openwhisk');
+const request = require('request')
+const { Kafka, logLevel } = require('kafkajs')
+
+
+// TODO combine query for performance
+async function loadFromCouchDB(id) {
+  const url = "https://2ce3adea-3579-4bbf-a74b-bfdc057c2806-bluemix.cloudant.com/images"
+
+  return new Promise((resolve, reject) => {
+    request.get({
+      url: url + "/" + id,
+      json: true,
+      // body:
+      auth: {
+        user: "fichavedellyoundbirstemp",
+        pass: "417164bcdc887b769c282f6d866ba3c5f62c4bda",
+        sendImmediately: true
+      }
+    }, (e, r, json) => {resolve(json)})
+  })
+}
+
+async function writeToKafka(topic, key, json) {
+  const kafka = new Kafka({
+    clientId: 'enhancer',
+    brokers: ['kafka03-prod02.messagehub.services.eu-gb.bluemix.net:9093', 'kafka02-prod02.messagehub.services.eu-gb.bluemix.net:9093', 'kafka01-prod02.messagehub.services.eu-gb.bluemix.net:9093'],
+    ssl: true,
+    sasl: {
+      mechanism: 'plain',
+      username: 'urGiSyjzBjRRNAHE',
+      password: 'rHICukLEqbSb9DIKrYmejTRRkMgDUkZK'
+    }
+  });
+
+  const producer = kafka.producer()
+  await producer.connect()
+  await producer.send({
+    topic: topic,
+    messages: [{
+      key: key,
+      value: json
+    }]
+  })
+  await producer.disconnect()
+}
+
+async function main() {
+  // TODO: remove generated images from CouchDB
+
+  // Submit all images to kafka
+  let img1 = await loadFromCouchDB("1.jpg")
+  // console.log(img1)
+  writeToKafka("topic1", "1.jpg", JSON.stringify(img1)).catch(e => console.error(e))
+  let start = new Date()
+
+  // TODO: wait until all images are in CouchDB
+  let end = new Date()
+
+  return {
+    headers: { 'Content-Type': 'application/json' },
+    body: { duration: end - start }
+  }
+}
+
+exports.main = main

kafka-images/schedulor/package.json

+{
+  "name": "schedulor",
+  "main": "index.js",
+  "dependencies": {
+    "kafkajs": "*"
+  }
+}