master merge to typescript

Author: gspletzer

.eslintrc.js

@@ -81,10 +81,13 @@ module.exports = {
   ],
   rules: {
     "arrow-body-style": 0,
+    camelcase: 0,
     "class-methods-use-this": 0,
+    "consistent-return": 0,
     "comma-dangle": 0,
     "dot-notation": 0,
     "func-names": 0,
+    "guard-for-in": 0,
     "import/extensions": 0,
     "import/no-extraneous-dependencies": 0,
     "import/no-unresolved": 0,
@@ -106,7 +109,7 @@ module.exports = {
     quotes: 0,
     "prefer-const": 1,
     "prefer-destructuring": 0,
-    "prefer-template": 1,
+    "prefer-template": 0,
     "react/button-has-type": 0,
     "react/destructuring-assignment": 0,
     "react/forbid-prop-types": 0,

SSEController.js

@@ -13,16 +13,20 @@ SSEController.createStream = (reqResObj, options, event) => {
   // because EventSource cannot access headers, we are making a regular get request to SSE server to get its headers, and then passing those headers into function where we will be connecting our EventSource, there will a time delay between the time the user opens the request and the server sends back its first response. We keep reference to the time the first request was made to account for that time difference later on. 
   const startTime = Date.now(); 
 
-  http.get(headers.url, {
+  const req = http.get(headers.url, {
     headers,
     agent: false,
-  }, (res) => {
-    // update info in reqResObj to reflect fact that connection was succesful
-    reqResObj.response.headers = {...res.headers};
-    reqResObj.connection = 'open'; 
-    reqResObj.connectionType = 'SSE';
-    // invoke function that will create an EventSource
-    SSEController.readStream(reqResObj, event, Date.now() - startTime);
+  });
+  req.once('response', (res) => {
+      // update info in reqResObj to reflect fact that connection was succesful
+      reqResObj.response.headers = {...res.headers};
+      reqResObj.connection = 'open'; 
+      reqResObj.connectionType = 'SSE';
+      // this is for purpose of logic in graph.jsx, which needs the entire req/res obj to have a timeReceived
+      reqResObj.timeReceived = Date.now();
+      // invoke function that will create an EventSource
+      SSEController.readStream(reqResObj, event, Date.now() - startTime);
+      req.destroy(); 
   });
 };
 
@@ -43,8 +47,9 @@ SSEController.readStream = (reqResObj, event, timeDiff) => {
     return event.sender.send('reqResUpdate', reqResObj);
   }; 
   sse.onerror = (err) => {
-    console.log('there was an error in SSEController.readStream', err)
-    };
+    console.log('there was an error in SSEController.readStream', err);
+    sse.close();
+  };
 };
 
 module.exports = SSEController; 

grpc_mockData/protos/helloworld.proto

@@ -1,14 +1,15 @@
-syntax = "proto3";
-package helloworld;
+// now refactored to work with testServer.js in the newer version of grpc (@grpc/grpc-js)
 
-service Greeter {
-  rpc SayHello (HelloRequest) returns (HelloReply) {}
+syntax= "proto3";
+
+service HelloWorldService {
+  rpc GreetMe (GreetRequest) returns (GreetReply) {}
 }
 
-message HelloRequest {
+message GreetRequest {
   string name = 1;
 }
 
-message HelloReply {
-  string message = 1;
+message GreetReply {
+  string reply = 1;
 }

grpc_mockData/protos/hw2.proto

@@ -7,8 +7,8 @@ service Greeter {
   // Sends a greeting
   rpc SayHello (HelloRequest) returns (HelloReply) {} // single unary stream
   rpc SayHelloNested (HelloNestedRequest) returns (HelloNestedReply) {} // nested unary stream
+  rpc SayHellosSs (HelloRequest) returns (stream HelloReply) {}
   rpc SayHelloCS (stream HelloRequest) returns (HelloReply) {}
-  rpc SayHellos (HelloRequest) returns (stream HelloReply) {}
   rpc SayHelloBidi (stream HelloRequest) returns (stream HelloReply) {}
 }
 

grpc_mockData/server.js

@@ -1,162 +1,106 @@
 const path = require("path");
-const Mali = require("mali");
-// consider replacing highland with normal node code for converting array to streams
+const fs = require("fs");
 const hl = require("highland");
+const Mali = require("mali");
+// Mali needs the old grpc as a peer dependency so that should be installed as well
 const grpc = require("@grpc/grpc-js");
 
+// consider replacing highland with normal node code for converting array to streams
+
 const PROTO_PATH = path.join(__dirname, "./protos/hw2.proto");
 const HOSTPORT = "0.0.0.0:50051";
 
-const dataStream = [
-  {
-    message: "You",
-  },
-  {
-    message: "Are",
-  },
-  {
-    message: "doing IT",
-  },
-  {
-    message: "Champ",
-  },
-];
-
-/**
- * Implements the SayHello RPC method.
- */
-
 // Unary stream
 // ctx = watch execution context
-function sayHello(ctx) {
-  // create new metadata
-  let metadata = new grpc.Metadata();
-  metadata.set("it", "works?");
-  metadata.set("indeed", "it do");
-  // Watcher creates a watch execution context for the watch
-  // The execution context provides scripts and templates with access to the watch metadata
-  // console.log("received metadata from client request", ctx.metadata)
-  // console.dir(ctx.metadata, { depth: 3, colors: true });
-  // console.log(`got sayHello request name: ${ctx.req.name}`);
-
-  // an alias to ctx.response.res
-  // This is set only in case of DUPLEX calls, to the the gRPC call reference itself
+async function sayHello(ctx) {
+  // ctx contains both req and res objects
+  // sets key-value pair inside ctx.response.metadata as a replacement for headers
+  ctx.set("UNARY", "true");
   ctx.res = { message: "Hello " + ctx.req.name };
-
-  // send response header metadata object directly as an argument and that is set and sent
-  metadata.set("UNARY", "yes");
-  ctx.sendMetadata(metadata);
-
-  // console.log(`set sayHello response: ${ctx.res.message}`);
 }
 // nested Unary stream
-
-function sayHelloNested(ctx) {
-  // create new metadata
-  let metadata = new grpc.Metadata();
-  metadata.set("it", "works?");
-  metadata.set("indeed", "it do");
-  // Watcher creates a watch execution context for the watch
-  // The execution context provides scripts and templates with access to the watch metadata
-  // console.log("received metadata from client request", ctx.metadata)
-  // console.dir(ctx.metadata, { depth: 3, colors: true });
-  // console.log("ctx line 64 from server.js", ctx)
-
+async function sayHelloNested(ctx) {
+  ctx.set("UNARY", "true");
   // nested unary response call
-  let firstPerson = ctx.req.firstPerson.name;
-  let secondPerson = ctx.req.secondPerson.name;
-  // console.log("firstPerson line 68 from server.js:", firstPerson)
+  const firstPerson = ctx.req.firstPerson.name;
+  const secondPerson = ctx.req.secondPerson.name;
   ctx.res = {
     serverMessage: [
       { message: "Hello! " + firstPerson },
       { message: "Hello! " + secondPerson },
     ],
   };
-
-  // send response header metadata object directly as an argument and that is set and sent
-  ctx.sendMetadata(metadata);
 }
 // Server-Side Stream
 // used highland library to manage asynchronous data
-async function sayHellos(ctx) {
-  // create new metadata
-  let metadata = new grpc.Metadata();
-  metadata.set("it", "works?");
-  metadata.set("indeed", "it do");
-  // The execution context provides scripts and templates with access to the watch metadata
-  // console.dir(ctx.metadata, { depth: 3, colors: true });
-  // converts a request into strings
-  // console.log(`got sayHellos request name:`, JSON.stringify(ctx.req, null, 4));
-
-  // alias for ctx.request.req
+async function sayHellosSs(ctx) {
+  ctx.set("Server-side-stream", "true");
   // In case of UNARY and RESPONSE_STREAM calls it is simply the gRPC call's request
 
-  let reqMessages = { message: "hello!!! " + ctx.req.name };
-
-  dataStream.push(reqMessages);
-  reqMessages = dataStream;
-  let streamData = await hl(reqMessages);
-  ctx.res = streamData;
-  metadata.set("serverStream", "indeed");
-  dataStream.pop();
-
-  // send response header metadata object directly as an argument and that is set and sent
-  ctx.sendMetadata(metadata);
-
+  const dataStream = [
+    {
+      message: "You",
+    },
+    {
+      message: "Are",
+    },
+    {
+      message: "doing IT",
+    },
+    {
+      message: "Champ",
+    },
+  ];
+
+  const reqMessages = { message: "hello!!! " + ctx.req.name };
+  // combine template with reqMessage
+  const updatedStream = [...dataStream, reqMessages];
+  const makeStreamData = await hl(updatedStream);
+  ctx.res = makeStreamData;
   // ends server stream
   ctx.res.end();
 }
 
 // Client-Side stream
-function sayHelloCs(ctx) {
+async function sayHelloCs(ctx) {
   // create new metadata
-  let metadata = new grpc.Metadata();
-  metadata.set("it", "works?");
-  metadata.set("indeed", "it do");
-  metadata.set("clientStream", "indubitably");
-  // The execution context provides scripts and templates with access to the watch metadata
-  console.dir(ctx.metadata, { depth: 3, colors: true });
-  // console.log('got sayHelloClients')
-  let counter = 0;
-  let messages = [];
-  // client streaming calls to write messages and end writing before you can get the response
+  ctx.set("client-side-stream", "true");
+
+  const messages = [];
+
   return new Promise((resolve, reject) => {
+    // ctx.req is the incoming readable stream
     hl(ctx.req)
       .map((message) => {
-        counter++;
-        // console.log('message content',message.name)
-        ctx.response.res = { message: "Client stream: " + message.name };
-        messages.push(message.name);
-        ctx.sendMetadata(metadata);
+        console.log("parsed stream message with name key, ", message);
+        // currently the proto file is setup to only read streams with the key "name"
+        // other named keys will be pushed as an empty object
+        messages.push(message);
+        return undefined;
       })
-      // returns all the elements as an array
       .collect()
       .toCallback((err, result) => {
         if (err) return reject(err);
-        // console.log(`done sayHelloClients counter ${counter}`)
-        ctx.response.res = { message: "SAYHELLOCs Client stream: " + messages };
-        // console.log(ctx.response.res)
-        resolve();
+        console.log("messages ->", messages);
+        ctx.response.res = { message: `received ${messages.length} messages` };
+        return resolve();
       });
   });
 }
 
 // Bi-Di stream
 function sayHelloBidi(ctx) {
   // create new metadata
-  let metadata = new grpc.Metadata();
-  metadata.set("it", "works?");
-  metadata.set("indeed", "it do");
-  // console.log("got sayHelloBidi");
+  ctx.set("bidi-stream", "true");
+  console.log("got sayHelloBidi");
   // The execution context provides scripts and templates with access to the watch metadata
   console.dir(ctx.metadata, { depth: 3, colors: true });
   let counter = 0;
-  ctx.req.on("data", (d) => {
+  ctx.req.on("data", (data) => {
     counter++;
-    ctx.res.write({ message: "bidi stream: " + d.name });
+    ctx.res.write({ message: "bidi stream: " + data.name });
   });
-  metadata.set("bidiStream", "ohyes");
-  ctx.sendMetadata(metadata);
+
   // calls end to client before closing server
   ctx.req.on("end", () => {
     // console.log(`done sayHelloBidi counter ${counter}`);
@@ -171,7 +115,7 @@ function sayHelloBidi(ctx) {
  */
 function main() {
   const app = new Mali(PROTO_PATH, "Greeter");
-  app.use({ sayHello, sayHelloNested, sayHellos, sayHelloCs, sayHelloBidi });
+  app.use({ sayHello, sayHelloNested, sayHellosSs, sayHelloCs, sayHelloBidi });
   app.start(HOSTPORT);
   console.log(`Greeter service running @ ${HOSTPORT}`);
 }

grpc_mockData/testServer.js

@@ -0,0 +1,25 @@
+// an example of making a test server without Mali and pure @grpc
+
+const path = require("path");
+const grpc = require("@grpc/grpc-js");
+const protoLoader = require("@grpc/proto-loader");
+
+// change PROTO_PATH to load a different mock proto file
+const PROTO_PATH = path.resolve(__dirname, "protos/helloworld.proto");
+const PORT = "0.0.0.0:50051";
+
+const proto = protoLoader.loadSync(PROTO_PATH);
+const definition = grpc.loadPackageDefinition(proto);
+
+const greetMe = (call, callback) => {
+  callback(null, { reply: `Hey ${call.request.name}!` });
+};
+
+const server = new grpc.Server();
+
+server.addService(definition.HelloWorldService.service, { greetMe });
+
+server.bindAsync(PORT, grpc.ServerCredentials.createInsecure(), (port) => {
+  server.start();
+  console.log(`grpc server running on port ${PORT}`);
+});