agent.go

package main

import (
	"encoding/json"
	"flag"
	"path/filepath"
	"time"

	nats "github.com/nats-io/nats.go"
	log "github.com/sirupsen/logrus"
	"prostocklivestock.com.au/thetooth/tagreader/internal/tag"

	"github.com/tarm/serial"
)

func main() {
	var urls = flag.String("s", "nats://192.168.0.246:4222,nats://192.168.0.245:4222,nats://192.168.0.244:4222", "The nats server URLs (separated by comma)")
	var subj = flag.String("subject", "v1.ring.reader", "Subject to publish tags to.")
	var port = flag.String("port", "/dev/ttyUSB0", "The _SERIAL_ port to use.")
	var tim1 = flag.Int("timeout", 1000, "Time in milliseconds before assuming reader has failed.")
	var tim2 = flag.Int("holdoff", 30, "Number of timeouts before assuming failure is critical.")

	flag.Parse()

	log.SetFormatter(&log.JSONFormatter{})

	// Connect to NATs
	nc, err := nats.Connect(*urls, nats.Name("reader-agent"), nats.DisconnectErrHandler(func(_ *nats.Conn, err error) {
		log.Fatalf("Client disconnected: %v", err)
	}))
	if err != nil {
		log.Fatalf("Can't connect: %v\n", err)
	}
	defer nc.Close()

	// Connect to serial port
	c := &serial.Config{Name: *port, Baud: 9600, Parity: serial.ParityNone, StopBits: 1}
	s, err := serial.OpenPort(c)
	if err != nil {
		log.Fatal(err)
	}
	defer s.Close()

	// Perform initial set-up of the reader.
	initReader(s, time.Millisecond*time.Duration(*tim1))

	// Reset handler
	if _, err := nc.Subscribe(*subj+".reset", func(msg *nats.Msg) {
		log.Infof("Got reset, reconfiguring S251B at %s", *port)
		if _, err := s.Write([]byte("L\n\r")); err != nil {
			log.Error(err)
		}
	}); err != nil {
		log.Error(err)
	}

	st := make(chan string)
	go func() {
		buf := make([]byte, 1)
		for {
			_, err := s.Read(buf)
			if err != nil {
				log.Fatal(err)
			}
			st <- string(buf)
		}
	}()

	// Reader loop. Scan port at 1 byte per pass, when we identify 'L'
	// character clear tag value and start appending. When encountering
	// a carriage return '\r', check if the contents of tag constitutes
	// a valid TIRIS Animal Format.
	// See: http://www.ti.com/lit/ug/scbu028/scbu028.pdf

	tagbuf := ""
	timeout := time.NewTicker(time.Millisecond * time.Duration(*tim1))
	timeoutCount := 0
	for {
		select {
		case symbol := <-st:
			switch symbol {
			case "L":
				tagbuf = ""
			case "\r":
				if len(tagbuf) < 27 || tagbuf[:2] != "LA" {
					if len(tagbuf) > 27 && tagbuf[:1] == "L" && tagbuf[3:4] == "A" {
						log.Debug("Special reader")
					} else {
						log.Debug("Invalid tag")
						continue
					}
				}

				// Comply with NLIS format
				tagbuf = tagbuf[len(tagbuf)-16:]

				log.Printf("Port: %s Tag: %s", *port, tagbuf)

				pack := tag.Payload{Tag: tagbuf, ReaderName: filepath.Base(*port)}

				b, err := json.Marshal(pack)
				if err != nil {
					log.Fatal(err)
					return
				}

				if err := nc.Publish(*subj+".stream", b); err != nil {
					log.Fatal(err)
					return
				}
			}

			// Append symbol
			tagbuf = tagbuf + symbol

			// Reset timeout on symbol
			timeout.Stop()
			timeout = time.NewTicker(time.Millisecond * time.Duration(*tim1))
			timeoutCount = 0
		case <-timeout.C:
			log.Warnf("Timer expired after %d ms waiting for a symbol, reconfiguring S251B at %s", *tim1, *port)
			initReader(s, time.Millisecond*time.Duration(*tim1))

			if timeoutCount >= *tim2 {
				log.Fatalf("Timeout occured %d times in a row! Going down!!!", timeoutCount)
			}

			timeoutCount++
		}
	}
}

func initReader(s *serial.Port, timeout time.Duration) {
	t := time.NewTimer(timeout)
	c := make(chan struct{}, 1)
	go func() {
		if _, err := s.Write([]byte("L\n\r")); err != nil {
			log.Error(err)
		}
		c <- struct{}{}
	}()
	select {
	case <-t.C:
		log.Fatal("Writing to port timed out!")
	case <-c:
		return
	}
}