From f6c7a7595ad94d80879c027eed8826bd629c2bc9 Mon Sep 17 00:00:00 2001 From: Nonoo Date: Sun, 25 Oct 2020 16:15:39 +0100 Subject: [PATCH] Add TCP server for rigctl --- args.go | 3 ++ serialstream.go | 25 ++++++++-- serialtcpsrv.go | 128 ++++++++++++++++++++++++++++++++++++++++++++++++ streamcommon.go | 4 +- 4 files changed, 155 insertions(+), 5 deletions(-) create mode 100644 serialtcpsrv.go diff --git a/args.go b/args.go index f556585..acb6707 100644 --- a/args.go +++ b/args.go @@ -7,10 +7,12 @@ import ( ) var connectAddress string +var serialTCPPort uint16 func parseArgs() { h := getopt.BoolLong("help", 'h', "display help") a := getopt.StringLong("address", 'a', "IC-705", "Connect to address") + t := getopt.Uint16Long("serial-tcp-port", 'p', 4532, "Expose serial port as TCP port for rigctl") getopt.Parse() @@ -20,4 +22,5 @@ func parseArgs() { } connectAddress = *a + serialTCPPort = *t } diff --git a/serialstream.go b/serialstream.go index 202edab..837a59f 100644 --- a/serialstream.go +++ b/serialstream.go @@ -13,6 +13,7 @@ type serialStream struct { common streamCommon serialPort serialPortStruct + tcpsrv serialTCPSrv sendSeq uint16 @@ -69,13 +70,20 @@ func (s *serialStream) sendOpenClose(close bool) error { func (s *serialStream) handleRead(r []byte) { if len(r) >= 22 { if r[16] == 0xc1 && r[0]-0x15 == r[17] { - s.serialPort.write <- r[21:] + r = r[21:] + + // log.Print("fromradio ", r) + + s.serialPort.write <- r + if s.tcpsrv.toClient != nil { + s.tcpsrv.toClient <- r + } } } } -func (s *serialStream) gotDataFromSerialPort(r []byte) { +func (s *serialStream) gotDataForRadio(r []byte) { for len(r) > 0 && !s.readFromSerialPort.frameStarted { if s.readFromSerialPort.buf.Len() > 1 { s.readFromSerialPort.buf.Reset() @@ -109,6 +117,8 @@ func (s *serialStream) gotDataFromSerialPort(r []byte) { for _, b := range r { s.readFromSerialPort.buf.WriteByte(b) if b == 0xfc || b == 0xfd || s.readFromSerialPort.buf.Len() == maxSerialFrameLength { + // log.Print("toradio ", s.readFromSerialPort.buf.Bytes()) + if err := s.send(s.readFromSerialPort.buf.Bytes()); err != nil { reportError(err) } @@ -128,7 +138,9 @@ func (s *serialStream) loop() { case r := <-s.common.readChan: s.handleRead(r) case r := <-s.serialPort.read: - s.gotDataFromSerialPort(r) + s.gotDataForRadio(r) + case r := <-s.tcpsrv.fromClient: + s.gotDataForRadio(r) case <-s.readFromSerialPort.frameTimeout.C: s.readFromSerialPort.buf.Reset() s.readFromSerialPort.frameStarted = false @@ -166,6 +178,10 @@ func (s *serialStream) start(devName string) error { return err } + if err := s.tcpsrv.start(); err != nil { + return err + } + s.deinitNeededChan = make(chan bool) s.deinitFinishedChan = make(chan bool) @@ -184,10 +200,11 @@ func (s *serialStream) init() error { } func (s *serialStream) deinit() { - if s.common.pkt0.sendTicker != nil { // Stream opened? + if s.common.conn != nil { _ = s.sendOpenClose(true) } + s.tcpsrv.stop() s.serialPort.deinit() if s.deinitNeededChan != nil { diff --git a/serialtcpsrv.go b/serialtcpsrv.go new file mode 100644 index 0000000..bb27a1a --- /dev/null +++ b/serialtcpsrv.go @@ -0,0 +1,128 @@ +package main + +import ( + "fmt" + "io" + "net" + + "github.com/nonoo/kappanhang/log" +) + +type serialTCPSrv struct { + listener net.Listener + client net.Conn + + fromClient chan []byte + toClient chan []byte + + writeLoopDeinitNeededChan chan bool + writeLoopDeinitFinishedChan chan bool + + deinitFinishedChan chan bool +} + +func (s *serialTCPSrv) writeLoop(errChan chan error) { + s.toClient = make(chan []byte) + defer func() { + s.toClient = nil + }() + + var b []byte + for { + select { + case b = <-s.toClient: + case <-s.writeLoopDeinitNeededChan: + s.writeLoopDeinitFinishedChan <- true + return + } + + for len(b) > 0 { + written, err := s.client.Write(b) + if err != nil { + errChan <- err + } + b = b[written:] + } + } +} + +func (s *serialTCPSrv) disconnectClient() { + if s.client != nil { + s.client.Close() + } + if s.writeLoopDeinitNeededChan != nil { + s.writeLoopDeinitNeededChan <- true + <-s.writeLoopDeinitFinishedChan + + s.writeLoopDeinitNeededChan = nil + s.writeLoopDeinitFinishedChan = nil + } +} + +func (s *serialTCPSrv) loop() { + for { + var err error + s.client, err = s.listener.Accept() + + if err != nil { + if err != io.EOF { + reportError(err) + } + s.listener.Close() + s.disconnectClient() + s.deinitFinishedChan <- true + return + } + + log.Print("client ", s.client.RemoteAddr().String(), " connected") + + s.writeLoopDeinitNeededChan = make(chan bool) + s.writeLoopDeinitFinishedChan = make(chan bool) + writeErrChan := make(chan error) + go s.writeLoop(writeErrChan) + + b := make([]byte, maxSerialFrameLength) + connected := true + for connected { + n, err := s.client.Read(b) + if err != nil { + break + } + + select { + case s.fromClient <- b[:n]: + case <-writeErrChan: + connected = false + } + } + + s.client.Close() + log.Print("client ", s.client.RemoteAddr().String(), " disconnected") + } +} + +func (s *serialTCPSrv) start() (err error) { + s.listener, err = net.Listen("tcp", fmt.Sprint(":", serialTCPPort)) + if err != nil { + fmt.Println(err) + return + } + + log.Print("exposing serial port on tcp port ", serialTCPPort) + + s.fromClient = make(chan []byte) + + s.deinitFinishedChan = make(chan bool) + go s.loop() + return +} + +func (s *serialTCPSrv) stop() { + if s.listener != nil { + s.listener.Close() + } + + s.disconnectClient() + close(s.fromClient) + <-s.deinitFinishedChan +} diff --git a/streamcommon.go b/streamcommon.go index ecbcbf6..0343973 100644 --- a/streamcommon.go +++ b/streamcommon.go @@ -176,7 +176,9 @@ func (s *streamCommon) deinit() { if s.gotRemoteSID && s.conn != nil { _ = s.sendDisconnect() } - s.conn.Close() + if s.conn != nil { + s.conn.Close() + } if s.readerCloseNeededChan != nil { s.readerCloseNeededChan <- true