- Add "Queue.js"

- Add sequelize and sqlite
- Add "App" model
- Add "register" operation
- Moved queue from Array to Queue object

.gitignore

@@ -1,4 +1,5 @@
 dst
+*.sqlite
 # Created by https://www.gitignore.io/api/node
 
 ### Node ###

package.json

@@ -22,6 +22,8 @@
     "nodemon": "^1.18.3"
   },
   "dependencies": {
-    "chalk": "^2.4.1"
+    "chalk": "^2.4.1",
+    "sequelize": "^4.38.0",
+    "sqlite3": "^4.0.2"
   }
 }

src/connectionHandler.js

@@ -3,18 +3,23 @@
 import type { Socket } from "net";
 
 const countAndSlice = require("./utils/countAndSlice");
+const Queue = require("./utils/Queue");
+const db = require("./db");
 
 // All connections stored by id
 const Connections = {};
 
 // List of operations should be executed
-const secureOp = ["ping", "close"];
+const secureOp = [
+  "register",
+  "ping",
+  "close"];
 
 class Connection {
   id: number;
   socket: Socket;
   buffer: string;
-  queue: string[];
+  queue: Queue<string>;
 
   constructor(id: number, socket: Socket) {
     console.log("New connection", id);
@@ -22,11 +27,11 @@ class Connection {
     this.id = id;
     this.socket = socket;
     this.buffer = "";
-    this.queue = [];
+    this.queue = new Queue();
 
     Connections[id] = this;
 
-    socket.on("data", data => {
+    socket.on("data", async data => {
       data = data.toString();
       this.buffer = this.buffer.concat(data);
       console.log("Buffer", this.buffer);
@@ -36,13 +41,12 @@ class Connection {
       } else {
         this.buffer = "";
       }
-      this.queue = this.queue.concat(arr);
+      this.queue.concat(arr);
 
       while (this.queue.length > 0) {
-        let op = this.queue.shift();
-        this.execute(op);
+        let op = this.queue.removeSync();
+        await this.execute(op);
       }
-
     })
 
     socket.on("end", () => {
@@ -50,10 +54,10 @@ class Connection {
     });
   }
 
-  execute(op: string) {
+  async execute(op: string) {
     if (secureOp.includes(op)) {
       // $FlowFixMe
-      this[op]();
+      await this[op]();
     } else {
       this.write("Unkown operator\n\r");
     }
@@ -61,12 +65,20 @@ class Connection {
 
   // Session`s Operations
 
+  async register() {
+    const name = await this.queue.remove();
+    let instance = await db.App.create({
+      name
+    });
+    this.write(`${instance.id};`);
+  }
+
   ping() {
-    this.write("pong\n\r");
+    this.write("pong;");
   }
 
   close() {
-    this.write("Closing connection\n\r");
+    this.write("Closing connection;");
     this.socket.end();
   }
 

src/db/App.js

@@ -0,0 +1,16 @@
+//@flow
+
+const sequelize = require("./sequelize");
+const Sequelize = require("sequelize");
+
+const App = sequelize.define("app", {
+  name: {
+    type: Sequelize.STRING
+  },
+  activated: {
+    type: Sequelize.BOOLEAN,
+    default: false
+  }
+});
+
+module.exports = App;

src/db/index.js

@@ -0,0 +1,12 @@
+//@flow
+
+const sequelize = require("./sequelize");
+
+const models = {
+  App: require("./App.js")
+};
+
+module.exports = {
+  sequelize,
+  ...models
+}

src/db/sequelize.js

@@ -0,0 +1,11 @@
+//@flow
+
+const Sequelize = require('sequelize');
+const sequelize = new Sequelize({
+  dialect: 'sqlite',
+  operatorsAliases: false,
+  storage: 'database.sqlite',
+  logging: false
+});
+
+module.exports = sequelize;

src/index.js

@@ -6,24 +6,31 @@ const chalk = require("chalk");
 
 const socketPathCon = require("./utils/socketPath");
 const connectionHandler = require("./connectionHandler");
+const db = require("./db");
 
 const server = net.createServer(connectionHandler);
 
 const socketPath = socketPathCon();
 
-server.listen(path.resolve(socketPath), () => {
-  console.log(`Server is listening on ${socketPath}`);
-});
-
-server.on("error", err => {
-  if (err.code == "EADDRINUSE") {
-    console.log(chalk.red(
-      `Totify stream address is already in use. ` +
-      `If you want run more instances, you should` +
-      ` give instance name in config of new instances`
-    ));
-  } else {
-    console.log("Unexpected error, code:", err.code);
-  }
-  process.exit();
-});
+//Main
+
+(async () => {
+  await db.sequelize.sync();
+
+  server.listen(path.resolve(socketPath), () => {
+    console.log(`Server is listening on ${socketPath}`);
+  });
+
+  server.on("error", err => {
+    if (err.code == "EADDRINUSE") {
+      console.log(chalk.red(
+        `Totify stream address is already in use. ` +
+        `If you want run more instances, you should` +
+        ` give instance name in config of new instances`
+      ));
+    } else {
+      console.log("Unexpected error, code:", err.code);
+    }
+    process.exit();
+  });
+})();

src/utils/Queue.js

@@ -0,0 +1,46 @@
+//@flow
+
+const ee = require("events");
+
+module.exports = class Queue<T> extends ee {
+  queue: Array<T>;
+
+  constructor() {
+    super();
+    this.queue = [];
+  }
+
+  get length(): number {
+    return this.queue.length;
+  }
+
+  add(el: T) {
+    this.queue.push(el);
+    this.emit("add");
+  }
+
+  remove(): Promise<T> {
+    return new Promise((resolve, reject) => {
+      if (this.queue.length > 0) {
+        this.emit("remove");
+        resolve(this.queue.shift());
+      } else {
+        this.once("add", () => {
+          resolve(this.removeSync());
+        })
+      }
+    });
+  }
+
+  removeSync(): T {
+    this.emit("remove");
+    return this.queue.shift();
+  }
+
+  concat(arr: Array<T>) {
+    if (arr.length > 0) {
+      this.queue = this.queue.concat(arr);
+      this.emit("add");
+    }
+  }
+}