kabu STATION API の PUSH API（時価PUSH配信）を記録し、あとから再生したい 06

昨日の続きをやる。結局、簡略化のためにRedisを挟まずダイレクトにDBに挿入することにした。

how-to-make-stock-trading-system.dogwood008.com

未だ動かないけど、こんな感じで実装している。

'use strict';

// ref: https://shizenkarasuzon.hatenablog.com/entry/2021/04/21/004132

import WebSocket from 'ws';

// ref: https://stackoverflow.com/q/70306590/15983717
import pg from 'pg';
const { Pool } = pg;

function initWebSocket () {
  const protocol = process.env.PROTOCOL;
  const host = process.env.HOST;
  const port = process.env.PORT;
  const path = process.env.WEBSOCKET_PATH || '/';
  const server = `${protocol}://${host}:${port}${path}`;

  console.log(server)
  return new WebSocket(server);
}

function createDbSql(dbName) {
  return `SELECT 'CREATE DATABASE ${dbName}'
    WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '${dbName}';`
}

function createTableSql(stockCode) {
  return `CREATE TABLE IF NOT EXISTS stock_${stockCode} (
    datetime timestamp NOT NULL,
    volume int NOT NULL,
    price float NOT NULL;`
}

function insertSql({ stock_code, dateTime, volume, price }) {
  return `INSERT INTO stock_${stock_code} VALUES(
    ${dateTime},
    ${volume},
    ${price}
  );`
}

async function initPg() {
  const pool = new Pool();
  const connect = await pool.connect();
  await connect.query(createDbSql('websocket_messages'))
  ['7974'].forEach(async stockCode => {
    await connect.query(createTableSql(stockCode))
  });

  return { pool, connect };
}

async function exit() {
  await client.end();
  await connect.release();
  await pool.end();
}

function main() {
  ws.on('open', function open() {
    console.log('open');
  });
  
  ws.on('error', async function (event) {
    console.log(event);
    await exit();
  });
  
  ws.on('close', async function close() {
    await exit();
  });

  ws.on('message', function message(data) {
    console.log('%s', data);
  });
}

process.on('SIGINT', function() {
  console.log('SIGINT');
  exit();
})

const ws = initWebSocket();
const { pool, connect } = initPg();
main();