Node Label Print Server for WASP WPL305

printserver.js

import Rx from "rx"
import Printer from "zuzel-printer"
import mysql from "mysql"

main()

function main () {
  let printer = new Printer("labels")
  let db = mysql.createConnection({
    host     : "abc",
    user     : "xxx",
    password : "zzz",
    database : "munchbox"
  })
  db.connect()

  let is_disposing = false
  let dispose = () => {
    if(is_disposing) return
    is_disposing = true
    printer.destroy()
    db.end()
  }

  process.on("exit", dispose)
  process.on("SIGINT", dispose)

  let set_label_as_printed = label =>
    query("update label_queue set status = 'printed' where id = " + label.id, db)
      .retryWhen(errors => errors.delay(10000))

  Rx.Observable
    .interval(3000)
    .timeInterval()
    .flatMap(query_labels(db))
    .flatMap(labels => create_print_jobs(labels, printer))
    .tap(job => console.log(job))
    .filter(job => job.completed)
    .flatMap(job => job.labels)
    .flatMap(set_label_as_printed)
    .subscribe()

  console.log("Printer server running")
}

function query(sql, db) {
  return Rx.Observable.create(observer => {
    db.query(sql, (err, rows) => {
      if(err) return observer.onError(err)
      observer.onNext(rows)
      observer.onCompleted()
    })
  })
}

function query_labels (db) {
  let labels_to_ids = labels => labels.map(label => label.id).join(",")
  let pending = false
  return Rx.Observable.create(observer => {
    if(pending) return
    pending = true
    query("select id, data from label_queue where status = 'pending'", db)
      .flatMap(labels =>
        labels.length
          ? query("update label_queue set status = 'processing' where id in (" + labels_to_ids(labels) + ")", db)
              .map(() => labels)
          : Rx.Observable.just([]))
      .retryWhen(errors => errors.delay(10000))
      .subscribe(labels => {
        if(labels.length)
          observer.onNext(labels)
        pending = false
      })
  })
}

function create_print_jobs (labels, printer) {
  if(!labels.length) return Rx.Observable.just({ completed: true, labels: [] })
  // Perhaps the data is streamed too fast, but unless there's a bunch of newlines, the next time labels are printed,
  // some of the last labels are going to be capped.
  let newlines = "\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n"
  // Labels are joined to enable fastest possible printing. There will be a long delay between every label otherwise.
  let data = labels.map(label => label.data).join(newlines) + newlines
  let job = printer.printBuffer(data, { o: "raw" })
  let sent = Rx.Observable.fromEvent(job, "sent").map(() => { return { sent: true, labels: labels } })
  let completed = Rx.Observable.fromEvent(job, "completed").map(() => { return { completed: true, labels: labels } })
  return Rx.Observable.merge(sent, completed)
}