roc/transport.go at master · abferm/roc · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
package roc

import (
	"io"
	"sync"

	"fmt"
	"net"
	"time"

	"github.com/abferm/serial"
)

type Transport interface {
	Transceive(request Message) (response Message, err error)
	Broadcast(request Message) (err error)
}

type baseTransport struct {
}

func (client baseTransport) send(request Message, transport io.Writer) (err error) {
	err = request.updateCalculatedFields()
	if err != nil {
		return
	}

	logger.Debugf("sending % x\n", request.bytes())
	_, err = transport.Write(request.bytes())
	if err != nil {
		return
	}
	return
}

func (client baseTransport) receive(transport io.Reader) (response Message, err error) {
	err = response.read(transport)
	if err != nil {
		return
	}
	if response.Opcode == ErrorResponse {
		var errM ErrorResponseMessage
		errM, err = errorResponseMessage(response)
		if err == nil {
			err = errM
		}
	}
	return
}

func (client baseTransport) transceive(request Message, transport io.ReadWriter) (response Message, err error) {
	err = client.send(request, transport)
	if err != nil {
		return
	}

	response, err = client.receive(transport)
	return
}

type SerialTransport struct {
	baseTransport
	Port   *serial.LockingSerialPort
	Config serial.Config
}

func NewSerialTransport(port *serial.LockingSerialPort, config serial.Config) *SerialTransport {
	trans := new(SerialTransport)
	trans.Port = port
	trans.Config = config
	return trans
}

func (trans SerialTransport) Transceive(request Message) (response Message, err error) {
	trans.Port.Lock()
	defer trans.Port.Unlock()
	err = trans.Port.Connect(trans.Config)
	if err != nil {
		return
	}
	defer trans.Port.Close()

	logger.Debugf("flushing serial port\n")
	if err = trans.Port.Flush(); err != nil {
		return
	}

	response, err = trans.transceive(request, trans.Port)
	return
}

func (trans SerialTransport) Broadcast(request Message) (err error) {
	trans.Port.Lock()
	defer trans.Port.Unlock()
	err = trans.Port.Connect(trans.Config)
	if err != nil {
		return
	}
	defer trans.Port.Close()

	logger.Debugf("flushing serial port\n")
	if err = trans.Port.Flush(); err != nil {
		return
	}

	err = trans.send(request, trans.Port)
	if err != nil {
		return
	}

	// Sleep 1/2 timeout in order to ensure the message makes it out before the
	// next flush
	time.Sleep(trans.Config.Timeout / 2)
	return
}

type TCPTransport struct {
	baseTransport
	Address string
	Port    int
	Timeout time.Duration
}

func NewTCPTransport(address string, port int, timeout time.Duration) *TCPTransport {
	trans := new(TCPTransport)
	trans.Address = address
	trans.Port = port
	trans.Timeout = timeout
	return trans
}

func (trans TCPTransport) Transceive(request Message) (response Message, err error) {
	conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", trans.Address, trans.Port))
	if err != nil {
		return
	}
	defer conn.Close()
	err = conn.SetDeadline(time.Now().Add(trans.Timeout))
	if err != nil {
		return
	}
	response, err = trans.transceive(request, conn)
	return
}

func (trans TCPTransport) Broadcast(request Message) (err error) {
	conn, err := net.Dial("tcp", fmt.Sprintf("%s:%d", trans.Address, trans.Port))
	if err != nil {
		return
	}
	defer conn.Close()
	err = conn.SetDeadline(time.Now().Add(trans.Timeout))
	if err != nil {
		return
	}
	err = trans.send(request, conn)
	return
}

// SharedTCPTransport: This is a TCPTransport wrapped in a mutex lock. It should
// be used when multiple devices are available at the same address and port, for
// example several serial devices connected to a serial to tcp converter via RS-485
// or a radio network.
type SharedTCPTransport struct {
	TCPTransport
	lock *sync.Mutex
}

func NewSharedTCPTransport(address string, port int, timeout time.Duration) *SharedTCPTransport {
	trans := new(SharedTCPTransport)
	trans.Address = address
	trans.Port = port
	trans.Timeout = timeout
	trans.lock = new(sync.Mutex)
	return trans
}

func (trans SharedTCPTransport) Transceive(request Message) (response Message, err error) {
	trans.lock.Lock()
	response, err = trans.TCPTransport.Transceive(request)
	trans.lock.Unlock()
	return
}

func (trans SharedTCPTransport) Broadcast(request Message) (err error) {
	trans.lock.Lock()
	err = trans.TCPTransport.Broadcast(request)
	trans.lock.Unlock()
	return
}