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
|
package canbus
import (
"context"
"net"
"strings"
"time"
"go.einride.tech/can"
"go.einride.tech/can/pkg/socketcan"
)
const timeout = 1 * time.Second
type Bus struct {
conn net.Conn
receiver *socketcan.Receiver
transmitter *socketcan.Transmitter
tx <-chan txChan
rx <-chan can.Frame
rxErr <-chan error
cancel context.CancelFunc
}
type txChan struct {
request chan<- transmission
err <-chan error
}
type transmission struct {
context.Context
can.Frame
}
func Connect(dev string) (Bus, error) {
conn, err := socketcan.Dial("can", dev)
if err != nil {
return Bus{}, err
}
receiver := socketcan.NewReceiver(conn)
transmitter := socketcan.NewTransmitter(conn)
tx := make(chan txChan)
rx := make(chan can.Frame)
rxErr := make(chan error)
ctx, cancel := context.WithCancel(context.Background())
go transmit(ctx, transmitter, tx)
go receive(ctx, receiver, rx, rxErr)
return Bus{conn, receiver, transmitter, tx, rx, rxErr, cancel}, nil
}
func transmit(ctx context.Context, transmitter *socketcan.Transmitter, tx chan<- txChan) {
defer close(tx)
for {
reqc := make(chan transmission)
errc := make(chan error)
select {
case tx <- txChan{reqc, errc}:
req := <-reqc
errc <- req.transmit(transmitter)
close(reqc)
close(errc)
case <-ctx.Done():
close(reqc)
close(errc)
return
}
}
}
func (t transmission) transmit(transmitter *socketcan.Transmitter) error {
ctx, _ := context.WithTimeout(context.Background(), timeout)
for {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
err := transmitter.TransmitFrame(t.Context, t.Frame)
if err == nil {
return nil // success
} else if strings.HasSuffix(err.Error(), "no buffer space available") {
continue // retry
} else {
return err
}
}
}
func receive(ctx context.Context, receiver *socketcan.Receiver, rx chan<- can.Frame, rxErr chan<- error) {
defer close(rx)
defer close(rxErr)
for receiver.Receive() {
frame := receiver.Frame()
select {
case rx <- frame:
case <-ctx.Done():
return
}
}
rxErr <- receiver.Err()
}
func (b Bus) Close() {
b.cancel()
b.transmitter.Close()
b.receiver.Close()
b.conn.Close()
}
func (b Bus) Send(ctx context.Context, frame can.Frame) error {
c := <-b.tx
c.request <- transmission{ctx, frame}
return <-c.err
}
func (b Bus) Receive(ctx context.Context) (can.Frame, error) {
select {
case frame := <-b.rx:
return frame, nil
case <-ctx.Done():
return can.Frame{}, ctx.Err()
}
}
func (b Bus) TryReceive() (can.Frame, bool) {
select {
case frame := <-b.rx:
return frame, true
default:
return can.Frame{}, false
}
}
|