Martino Ferrari
606 lines
14 KiB
Go
606 lines
14 KiB
Go
// Package testca provides an in-process fake CA server for use in tests.
|
|
// It supports a small, fixed set of PVs and is compatible with the goca Client.
|
|
package testca
|
|
|
|
import (
|
|
"encoding/binary"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"net"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/uopi/goca/proto"
|
|
)
|
|
|
|
// PVSpec describes one test PV hosted by the fake server.
|
|
type PVSpec struct {
|
|
Name string
|
|
DBFType int // proto.DBF* constant
|
|
Count uint32 // element count (1 for scalars)
|
|
Value any // initial value
|
|
Access uint32 // proto.AccessRead | proto.AccessWrite
|
|
}
|
|
|
|
// Server is an in-process fake CA server. It listens on a random TCP port and
|
|
// an ephemeral UDP port. Use Addr() to get the addresses for client config.
|
|
type Server struct {
|
|
tcpLn net.Listener
|
|
udpCn *net.UDPConn
|
|
|
|
mu sync.RWMutex
|
|
pvs map[string]*serverPV
|
|
subs map[uint32]*serverSub // subID → sub
|
|
|
|
done chan struct{}
|
|
}
|
|
|
|
type serverPV struct {
|
|
spec PVSpec
|
|
mu sync.RWMutex
|
|
val any
|
|
}
|
|
|
|
type serverSub struct {
|
|
pvName string
|
|
subID uint32
|
|
cid uint32
|
|
sid uint32
|
|
conn net.Conn
|
|
dbrType uint16
|
|
count uint32
|
|
}
|
|
|
|
// New creates and starts a fake CA server hosting the given PVs.
|
|
func New(pvs []PVSpec) (*Server, error) {
|
|
tcpLn, err := net.Listen("tcp4", "127.0.0.1:0")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("testca: tcp listen: %w", err)
|
|
}
|
|
udpCn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")})
|
|
if err != nil {
|
|
tcpLn.Close()
|
|
return nil, fmt.Errorf("testca: udp listen: %w", err)
|
|
}
|
|
|
|
s := &Server{
|
|
tcpLn: tcpLn,
|
|
udpCn: udpCn,
|
|
pvs: make(map[string]*serverPV, len(pvs)),
|
|
subs: make(map[uint32]*serverSub),
|
|
done: make(chan struct{}),
|
|
}
|
|
for _, spec := range pvs {
|
|
s.pvs[spec.Name] = &serverPV{spec: spec, val: spec.Value}
|
|
}
|
|
|
|
go s.acceptLoop()
|
|
go s.udpLoop()
|
|
return s, nil
|
|
}
|
|
|
|
// TCPAddr returns the TCP "host:port" address clients should connect to.
|
|
func (s *Server) TCPAddr() string { return s.tcpLn.Addr().String() }
|
|
|
|
// UDPAddr returns the UDP "host:port" address clients should search against.
|
|
func (s *Server) UDPAddr() string { return s.udpCn.LocalAddr().String() }
|
|
|
|
// Close shuts down the server.
|
|
func (s *Server) Close() {
|
|
close(s.done)
|
|
s.tcpLn.Close()
|
|
s.udpCn.Close()
|
|
}
|
|
|
|
// SetValue updates a PV's value and pushes a monitor event to all subscribers.
|
|
func (s *Server) SetValue(pvName string, val any) error {
|
|
s.mu.RLock()
|
|
pv, ok := s.pvs[pvName]
|
|
s.mu.RUnlock()
|
|
if !ok {
|
|
return fmt.Errorf("testca: unknown PV %q", pvName)
|
|
}
|
|
pv.mu.Lock()
|
|
pv.val = val
|
|
pv.mu.Unlock()
|
|
|
|
// Push to all subscribers of this PV.
|
|
s.mu.RLock()
|
|
for _, sub := range s.subs {
|
|
if sub.pvName == pvName {
|
|
if msg := s.buildEventMsg(sub, pv); msg != nil {
|
|
_, _ = sub.conn.Write(msg)
|
|
}
|
|
}
|
|
}
|
|
s.mu.RUnlock()
|
|
return nil
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// UDP search loop //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
func (s *Server) udpLoop() {
|
|
buf := make([]byte, 65536)
|
|
for {
|
|
_ = s.udpCn.SetReadDeadline(time.Now().Add(200 * time.Millisecond))
|
|
n, src, err := s.udpCn.ReadFromUDP(buf)
|
|
if err != nil {
|
|
select {
|
|
case <-s.done:
|
|
return
|
|
default:
|
|
continue
|
|
}
|
|
}
|
|
s.handleUDP(buf[:n], src)
|
|
}
|
|
}
|
|
|
|
func (s *Server) handleUDP(data []byte, src *net.UDPAddr) {
|
|
for len(data) >= proto.HeaderSize {
|
|
hdr, n, err := proto.DecodeHeader(newBytesReader(data))
|
|
if err != nil {
|
|
return
|
|
}
|
|
payEnd := n + int(hdr.PayloadSize)
|
|
if payEnd > len(data) {
|
|
return
|
|
}
|
|
payload := data[n:payEnd]
|
|
data = data[payEnd:]
|
|
|
|
if hdr.Command != proto.CmdSearch {
|
|
continue
|
|
}
|
|
// Extract PV name from payload (null-terminated).
|
|
pvName := nullStr(payload)
|
|
|
|
s.mu.RLock()
|
|
_, ok := s.pvs[pvName]
|
|
s.mu.RUnlock()
|
|
if !ok {
|
|
continue
|
|
}
|
|
|
|
// Reply: data_type = TCP port, parameter2 = searchID.
|
|
tcpPort := s.tcpLn.Addr().(*net.TCPAddr).Port
|
|
// Use 0xFFFFFFFF IP to tell client to use sender's IP.
|
|
reply := buildSearchReply(hdr.Parameter2, tcpPort)
|
|
_, _ = s.udpCn.WriteToUDP(reply, src)
|
|
}
|
|
}
|
|
|
|
func buildSearchReply(searchID uint32, port int) []byte {
|
|
payload := []byte{0xFF, 0xFF, 0xFF, 0xFF, 0x00, 0x00, 0x00, 0x00}
|
|
h := proto.Header{
|
|
Command: proto.CmdSearch,
|
|
DataType: uint16(port),
|
|
Parameter1: proto.MinorVersion,
|
|
Parameter2: searchID,
|
|
}
|
|
return proto.BuildMessage(h, payload)
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// TCP accept + per-connection handler //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
func (s *Server) acceptLoop() {
|
|
for {
|
|
conn, err := s.tcpLn.Accept()
|
|
if err != nil {
|
|
select {
|
|
case <-s.done:
|
|
return
|
|
default:
|
|
continue
|
|
}
|
|
}
|
|
go s.handleConn(conn)
|
|
}
|
|
}
|
|
|
|
func (s *Server) handleConn(conn net.Conn) {
|
|
defer conn.Close()
|
|
|
|
// Track CID → (pvName, SID).
|
|
type chanInfo struct {
|
|
pvName string
|
|
sid uint32
|
|
}
|
|
sidSeq := uint32(1000)
|
|
channels := make(map[uint32]*chanInfo) // cid → info
|
|
|
|
for {
|
|
hdr, _, err := proto.DecodeHeader(conn)
|
|
if err != nil {
|
|
return
|
|
}
|
|
var payload []byte
|
|
if hdr.PayloadSize > 0 {
|
|
payload = make([]byte, hdr.PayloadSize)
|
|
if _, err = io.ReadFull(conn, payload); err != nil {
|
|
return
|
|
}
|
|
}
|
|
|
|
switch hdr.Command {
|
|
case proto.CmdVersion:
|
|
// Respond with VERSION.
|
|
reply := proto.BuildMessage(proto.Header{
|
|
Command: proto.CmdVersion,
|
|
DataCount: proto.MinorVersion,
|
|
}, nil)
|
|
conn.Write(reply)
|
|
|
|
case proto.CmdHostName, proto.CmdClientName:
|
|
// Ignore client identity.
|
|
|
|
case proto.CmdCreateChan:
|
|
cid := hdr.Parameter1
|
|
pvName := nullStr(payload)
|
|
|
|
s.mu.RLock()
|
|
pv, ok := s.pvs[pvName]
|
|
s.mu.RUnlock()
|
|
|
|
if !ok {
|
|
// CREATE_FAIL
|
|
fail := proto.BuildMessage(proto.Header{
|
|
Command: proto.CmdCreateFail,
|
|
Parameter1: cid,
|
|
}, nil)
|
|
conn.Write(fail)
|
|
continue
|
|
}
|
|
|
|
sid := sidSeq
|
|
sidSeq++
|
|
channels[cid] = &chanInfo{pvName: pvName, sid: sid}
|
|
|
|
// CREATE_CHAN reply: DataType=dbfType, DataCount=count, p1=cid, p2=sid.
|
|
pv.mu.RLock()
|
|
dbfType := pv.spec.DBFType
|
|
count := pv.spec.Count
|
|
pv.mu.RUnlock()
|
|
|
|
reply := proto.BuildMessage(proto.Header{
|
|
Command: proto.CmdCreateChan,
|
|
DataType: uint16(dbfType),
|
|
DataCount: count,
|
|
Parameter1: cid,
|
|
Parameter2: sid,
|
|
}, nil)
|
|
conn.Write(reply)
|
|
|
|
// ACCESS_RIGHTS: p1=cid, p2=access.
|
|
access := pv.spec.Access
|
|
if access == 0 {
|
|
access = proto.AccessRead | proto.AccessWrite
|
|
}
|
|
ar := proto.BuildMessage(proto.Header{
|
|
Command: proto.CmdAccessRights,
|
|
Parameter1: cid,
|
|
Parameter2: access,
|
|
}, nil)
|
|
conn.Write(ar)
|
|
|
|
case proto.CmdEventAdd:
|
|
// p1=SID (channel), p2=subscriptionID (client-assigned).
|
|
sid := hdr.Parameter1
|
|
subID := hdr.Parameter2
|
|
|
|
// Find channel by SID.
|
|
var pvName string
|
|
for _, info := range channels {
|
|
if info.sid == sid {
|
|
pvName = info.pvName
|
|
break
|
|
}
|
|
}
|
|
if pvName == "" {
|
|
continue
|
|
}
|
|
|
|
s.mu.Lock()
|
|
sub := &serverSub{
|
|
pvName: pvName,
|
|
subID: subID,
|
|
sid: sid,
|
|
conn: conn,
|
|
dbrType: hdr.DataType,
|
|
count: hdr.DataCount,
|
|
}
|
|
s.subs[subID] = sub
|
|
s.mu.Unlock()
|
|
|
|
// Send initial value.
|
|
s.mu.RLock()
|
|
pv := s.pvs[pvName]
|
|
s.mu.RUnlock()
|
|
if msg := s.buildEventMsg(sub, pv); msg != nil {
|
|
conn.Write(msg)
|
|
}
|
|
|
|
case proto.CmdEventCancel:
|
|
// p1=SID, p2=subscriptionID.
|
|
subID := hdr.Parameter2
|
|
s.mu.Lock()
|
|
delete(s.subs, subID)
|
|
s.mu.Unlock()
|
|
|
|
case proto.CmdReadNotify:
|
|
// p1=SID, p2=ioid (per CA spec).
|
|
sid := hdr.Parameter1
|
|
ioid := hdr.Parameter2
|
|
|
|
var pvName string
|
|
var cid uint32
|
|
for c, info := range channels {
|
|
if info.sid == sid {
|
|
pvName = info.pvName
|
|
cid = c
|
|
break
|
|
}
|
|
}
|
|
if pvName == "" {
|
|
continue
|
|
}
|
|
|
|
s.mu.RLock()
|
|
pv := s.pvs[pvName]
|
|
s.mu.RUnlock()
|
|
|
|
pv.mu.RLock()
|
|
val := pv.val
|
|
pv.mu.RUnlock()
|
|
|
|
var replyPayload []byte
|
|
switch hdr.DataType {
|
|
case proto.DBRCtrlDouble, proto.DBRCtrlLong, proto.DBRCtrlEnum, proto.DBRCtrlString,
|
|
proto.DBRCtrlFloat, proto.DBRCtrlShort, proto.DBRCtrlChar:
|
|
replyPayload = s.encodeCtrlValue(hdr.DataType, pv)
|
|
default:
|
|
replyPayload = s.encodeTimeValue(hdr.DataType, hdr.DataCount, val)
|
|
}
|
|
reply := proto.BuildMessage(proto.Header{
|
|
Command: proto.CmdReadNotify,
|
|
DataType: hdr.DataType,
|
|
DataCount: hdr.DataCount,
|
|
Parameter1: cid, // echo client channel ID (matches real EPICS IOC)
|
|
Parameter2: ioid, // echo ioid so client can match the reply
|
|
}, replyPayload)
|
|
conn.Write(reply)
|
|
|
|
case proto.CmdWrite:
|
|
// p1=SID, p2=0.
|
|
sid := hdr.Parameter1
|
|
var pvName string
|
|
for _, info := range channels {
|
|
if info.sid == sid {
|
|
pvName = info.pvName
|
|
break
|
|
}
|
|
}
|
|
if pvName == "" {
|
|
continue
|
|
}
|
|
s.mu.RLock()
|
|
pv := s.pvs[pvName]
|
|
s.mu.RUnlock()
|
|
|
|
val := decodeWritePayload(hdr.DataType, payload)
|
|
if val == nil {
|
|
continue
|
|
}
|
|
pv.mu.Lock()
|
|
pv.val = val
|
|
pv.mu.Unlock()
|
|
|
|
// Push update to all subscribers.
|
|
s.mu.RLock()
|
|
for _, sub := range s.subs {
|
|
if sub.pvName == pvName {
|
|
if msg := s.buildEventMsg(sub, pv); msg != nil {
|
|
sub.conn.Write(msg)
|
|
}
|
|
}
|
|
}
|
|
s.mu.RUnlock()
|
|
|
|
case proto.CmdEcho:
|
|
// No response needed (server echoes are optional).
|
|
}
|
|
}
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// Value encoding helpers //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
func (s *Server) buildEventMsg(sub *serverSub, pv *serverPV) []byte {
|
|
pv.mu.RLock()
|
|
val := pv.val
|
|
pv.mu.RUnlock()
|
|
payload := s.encodeTimeValue(sub.dbrType, sub.count, val)
|
|
if payload == nil {
|
|
return nil
|
|
}
|
|
return proto.BuildMessage(proto.Header{
|
|
Command: proto.CmdEventAdd,
|
|
DataType: sub.dbrType,
|
|
DataCount: sub.count,
|
|
Parameter1: 1, // ECA_NORMAL
|
|
Parameter2: sub.subID,
|
|
}, payload)
|
|
}
|
|
|
|
// encodeTimeValue builds a DBR_TIME_* payload for val.
|
|
func (s *Server) encodeTimeValue(dbrType uint16, _ uint32, val any) []byte {
|
|
now := time.Now().UTC()
|
|
sec := uint32(now.Unix() - 631152000) // EPICS epoch offset
|
|
nsec := uint32(now.Nanosecond())
|
|
|
|
hdr := make([]byte, 12)
|
|
// status=0, severity=0
|
|
binary.BigEndian.PutUint32(hdr[4:], sec)
|
|
binary.BigEndian.PutUint32(hdr[8:], nsec)
|
|
|
|
var body []byte
|
|
switch dbrType {
|
|
case proto.DBRTimeDouble:
|
|
body = make([]byte, 12) // 4 pad + 8 value
|
|
f := toF64(val)
|
|
binary.BigEndian.PutUint64(body[4:], math.Float64bits(f))
|
|
|
|
case proto.DBRTimeFloat:
|
|
body = make([]byte, 4)
|
|
binary.BigEndian.PutUint32(body, math.Float32bits(float32(toF64(val))))
|
|
|
|
case proto.DBRTimeLong:
|
|
body = make([]byte, 4)
|
|
binary.BigEndian.PutUint32(body, uint32(int32(toF64(val))))
|
|
|
|
case proto.DBRTimeShort, proto.DBRTimeEnum:
|
|
body = make([]byte, 4) // 2-byte RISC pad + 2-byte value
|
|
binary.BigEndian.PutUint16(body[2:], uint16(int16(toF64(val))))
|
|
|
|
case proto.DBRTimeChar:
|
|
body = make([]byte, 4) // RISC_pad0[0:2] + RISC_pad1[2] + value[3]
|
|
body[3] = byte(uint8(toF64(val)))
|
|
|
|
case proto.DBRTimeString:
|
|
body = make([]byte, 40)
|
|
if str, ok := val.(string); ok {
|
|
copy(body, str)
|
|
}
|
|
|
|
default:
|
|
return nil
|
|
}
|
|
|
|
return append(hdr, body...)
|
|
}
|
|
|
|
// encodeCtrlValue builds a DBR_CTRL_* payload for a GET reply.
|
|
// The fake server returns minimal but structurally correct payloads so that
|
|
// the client-side decode functions succeed.
|
|
func (s *Server) encodeCtrlValue(dbrType uint16, pv *serverPV) []byte {
|
|
pv.mu.RLock()
|
|
val := pv.val
|
|
access := pv.spec.Access
|
|
pv.mu.RUnlock()
|
|
|
|
switch dbrType {
|
|
case proto.DBRCtrlDouble: // 88 bytes
|
|
p := make([]byte, 88)
|
|
if access == proto.AccessRead {
|
|
// status=0, severity=0 (read-only signalled via ACCESS_RIGHTS, not here)
|
|
}
|
|
// units at [8:16] (empty string = "")
|
|
// value at [80:88]
|
|
binary.BigEndian.PutUint64(p[80:], math.Float64bits(toF64(val)))
|
|
return p
|
|
|
|
case proto.DBRCtrlFloat, proto.DBRCtrlShort, proto.DBRCtrlChar: // map to Long-style
|
|
// Treat as CtrlLong (48 bytes); caller asked for these types but our
|
|
// NativeCtrlType maps them to Double/Long anyway. Return a valid stub.
|
|
p := make([]byte, 48)
|
|
binary.BigEndian.PutUint32(p[44:], uint32(int32(toF64(val))))
|
|
return p
|
|
|
|
case proto.DBRCtrlLong: // 48 bytes
|
|
p := make([]byte, 48)
|
|
binary.BigEndian.PutUint32(p[44:], uint32(int32(toF64(val))))
|
|
return p
|
|
|
|
case proto.DBRCtrlEnum: // 424 bytes
|
|
p := make([]byte, 424)
|
|
// no_str at [4:6] = 0 (no enum strings in fake server)
|
|
binary.BigEndian.PutUint16(p[422:], uint16(int16(toF64(val))))
|
|
return p
|
|
|
|
case proto.DBRCtrlString: // 44 bytes: status(2)+severity(2)+value[40]
|
|
p := make([]byte, 44)
|
|
if str, ok := val.(string); ok {
|
|
copy(p[4:], str)
|
|
}
|
|
return p
|
|
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func decodeWritePayload(dbrType uint16, payload []byte) any {
|
|
switch dbrType {
|
|
case proto.DBRDouble:
|
|
if len(payload) < 8 {
|
|
return nil
|
|
}
|
|
return math.Float64frombits(binary.BigEndian.Uint64(payload))
|
|
case proto.DBRLong:
|
|
if len(payload) < 4 {
|
|
return nil
|
|
}
|
|
return int32(binary.BigEndian.Uint32(payload))
|
|
case proto.DBRShort, proto.DBREnum:
|
|
if len(payload) < 2 {
|
|
return nil
|
|
}
|
|
return int16(binary.BigEndian.Uint16(payload))
|
|
case proto.DBRString:
|
|
return nullStr(payload)
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func toF64(v any) float64 {
|
|
switch x := v.(type) {
|
|
case float64:
|
|
return x
|
|
case float32:
|
|
return float64(x)
|
|
case int:
|
|
return float64(x)
|
|
case int32:
|
|
return float64(x)
|
|
case int16:
|
|
return float64(x)
|
|
case int64:
|
|
return float64(x)
|
|
default:
|
|
return 0
|
|
}
|
|
}
|
|
|
|
func nullStr(b []byte) string {
|
|
for i, c := range b {
|
|
if c == 0 {
|
|
return string(b[:i])
|
|
}
|
|
}
|
|
return string(b)
|
|
}
|
|
|
|
// -------------------------------------------------------------------------- //
|
|
// Minimal io.Reader for proto.DecodeHeader //
|
|
// -------------------------------------------------------------------------- //
|
|
|
|
type bytesReader struct{ b []byte; i int }
|
|
|
|
func newBytesReader(b []byte) *bytesReader { return &bytesReader{b: b} }
|
|
|
|
func (r *bytesReader) Read(p []byte) (int, error) {
|
|
if r.i >= len(r.b) {
|
|
return 0, io.EOF
|
|
}
|
|
n := copy(p, r.b[r.i:])
|
|
r.i += n
|
|
return n, nil
|
|
}
|