diff --git a/.gitignore b/.gitignore index dc8567e..6e23df1 100644 --- a/.gitignore +++ b/.gitignore @@ -5,3 +5,10 @@ interfaces/ synthetic.json uopi.toml *.local.toml +resources +.claude +.github +.goreleaser.yaml +.iocsh_history +go.work.sum +*.log diff --git a/cmd/catools/main.go b/cmd/catools/main.go index 8e879b0..c288ae0 100644 --- a/cmd/catools/main.go +++ b/cmd/catools/main.go @@ -1,12 +1,12 @@ -// Command catools provides caget/caput/camonitor/cainfo equivalents using -// the pure-Go goca CA client library. +// Command catools provides CA (Channel Access) command-line utilities using +// the pure-Go goca library. // // Usage: // -// catools caget [pv...] # get current value(s) -// catools caput # write a value -// catools cainfo [pv...] # print metadata (units, limits, type) -// catools camonitor [pv...] # stream live updates (Ctrl-C to stop) +// catools get [pv...] Get current value(s) +// catools put Write a value +// catools monitor [pv...] Stream live updates (Ctrl-C to stop) +// catools info [pv...] Print metadata (type, units, limits) // // Environment: // @@ -30,13 +30,13 @@ import ( ) func usage() { - fmt.Fprintln(os.Stderr, `catools — pure-Go EPICS CA command-line tools + fmt.Fprintln(os.Stderr, `catools — EPICS Channel Access command-line tool -Commands: - caget [pv...] Get current value(s) - caput Write a value - cainfo [pv...] Print metadata (type, units, limits) - camonitor [pv...] Stream live updates (Ctrl-C to stop) +Usage: + catools get [pv...] Get current value(s) + catools put Write a value + catools monitor [pv...] Stream live updates (Ctrl-C to stop) + catools info [pv...] Print metadata (type, units, limits) Environment: EPICS_CA_ADDR_LIST Space-separated CA server addresses @@ -63,36 +63,36 @@ func main() { } switch cmd { - case "caget": + case "get": if len(args) == 0 { - fatalf("caget: no PV names specified") + fatalf("get: no PV names specified") } - runCaget(ctx, cli, args) - case "caput": + runGet(ctx, cli, args) + case "put": if len(args) < 2 { - fatalf("caput: usage: catools caput ") + fatalf("put: usage: catools put ") } - runCaput(ctx, cli, args[0], args[1]) - case "cainfo": + runPut(ctx, cli, args[0], args[1]) + case "info": if len(args) == 0 { - fatalf("cainfo: no PV names specified") + fatalf("info: no PV names specified") } - runCainfo(ctx, cli, args) - case "camonitor": + runInfo(ctx, cli, args) + case "monitor": if len(args) == 0 { - fatalf("camonitor: no PV names specified") + fatalf("monitor: no PV names specified") } - runCamonitor(ctx, cli, args) + runMonitor(ctx, cli, args) default: - fatalf("unknown command %q — use caget, caput, cainfo, or camonitor", cmd) + fatalf("unknown command %q — use get, put, monitor, or info", cmd) } } // -------------------------------------------------------------------------- // -// caget // +// get // // -------------------------------------------------------------------------- // -func runCaget(ctx context.Context, cli *ca.Client, pvs []string) { +func runGet(ctx context.Context, cli *ca.Client, pvs []string) { tctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() @@ -107,14 +107,14 @@ func runCaget(ctx context.Context, cli *ca.Client, pvs []string) { } // -------------------------------------------------------------------------- // -// caput // +// put // // -------------------------------------------------------------------------- // -func runCaput(ctx context.Context, cli *ca.Client, pv, valueStr string) { +func runPut(ctx context.Context, cli *ca.Client, pv, valueStr string) { tctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - // First get metadata to determine the right type. + // Get metadata to determine the right type. ci, err := cli.GetCtrl(tctx, pv) if err != nil { fatalf("%s: get metadata: %v", pv, err) @@ -125,22 +125,18 @@ func runCaput(ctx context.Context, cli *ca.Client, pv, valueStr string) { case proto.DBFString: value = valueStr case proto.DBFEnum: - // Try numeric first, then string. if n, err := strconv.ParseInt(valueStr, 10, 32); err == nil { value = int32(n) - } else { - // Search enum strings. - if ci.Enum != nil { - for i, s := range ci.Enum.Strings { - if strings.EqualFold(s, valueStr) { - value = int32(i) - break - } + } else if ci.Enum != nil { + for i, s := range ci.Enum.Strings { + if strings.EqualFold(s, valueStr) { + value = int32(i) + break } } - if value == nil { - fatalf("%s: unknown enum value %q", pv, valueStr) - } + } + if value == nil { + fatalf("%s: unknown enum value %q", pv, valueStr) } default: if n, err := strconv.ParseFloat(valueStr, 64); err == nil { @@ -156,20 +152,19 @@ func runCaput(ctx context.Context, cli *ca.Client, pv, valueStr string) { fatalf("%s: put: %v", pv, err) } - // Read back the new value. tv, err := cli.Get(tctx, pv) if err != nil { - fmt.Printf("Old: ?\nNew: %s (write sent, readback failed: %v)\n", valueStr, err) + fmt.Printf("%-30s %s (write sent, readback failed: %v)\n", pv, valueStr, err) return } fmt.Printf("%-30s %s\n", pv, formatValue(tv)) } // -------------------------------------------------------------------------- // -// cainfo // +// info // // -------------------------------------------------------------------------- // -func runCainfo(ctx context.Context, cli *ca.Client, pvs []string) { +func runInfo(ctx context.Context, cli *ca.Client, pvs []string) { tctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() @@ -220,10 +215,10 @@ func printInfo(pv string, ci ca.CtrlInfo) { } // -------------------------------------------------------------------------- // -// camonitor // +// monitor // // -------------------------------------------------------------------------- // -func runCamonitor(ctx context.Context, cli *ca.Client, pvs []string) { +func runMonitor(ctx context.Context, cli *ca.Client, pvs []string) { type entry struct { pv string ch chan proto.TimeValue @@ -249,8 +244,6 @@ func runCamonitor(ctx context.Context, cli *ca.Client, pvs []string) { return } - // Fan out: one goroutine per PV, print to stdout. - done := make(chan struct{}) for _, e := range entries { go func(e *entry) { for { @@ -269,8 +262,6 @@ func runCamonitor(ctx context.Context, cli *ca.Client, pvs []string) { } <-ctx.Done() - close(done) - for _, e := range entries { e.can() } @@ -283,14 +274,14 @@ func runCamonitor(ctx context.Context, cli *ca.Client, pvs []string) { func formatValue(tv proto.TimeValue) string { var val string switch { - case tv.Doubles != nil: + case tv.Str != "": + val = tv.Str + case len(tv.Doubles) > 1: parts := make([]string, len(tv.Doubles)) for i, v := range tv.Doubles { parts[i] = strconv.FormatFloat(v, 'g', -1, 64) } val = "[" + strings.Join(parts, " ") + "]" - case tv.Str != "": - val = tv.Str default: val = strconv.FormatFloat(tv.Double, 'g', -1, 64) } diff --git a/cmd/pvtools/main.go b/cmd/pvtools/main.go new file mode 100644 index 0000000..4c56838 --- /dev/null +++ b/cmd/pvtools/main.go @@ -0,0 +1,253 @@ +// Command pvtools provides PV Access command-line utilities using +// the pure-Go gopva library. +// +// Usage: +// +// pvtools get [pv...] Get current value(s) via PVA +// pvtools put Write a value via PVA +// pvtools monitor [pv...] Stream live updates (Ctrl-C to stop) +// pvtools info [pv...] Print PVA structure info +// +// Environment: +// +// EPICS_PVA_ADDR_LIST — space-separated PVA server addresses +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + "strconv" + "strings" + "syscall" + "time" + + pva "github.com/uopi/gopva" + "github.com/uopi/gopva/pvdata" +) + +func usage() { + fmt.Fprintln(os.Stderr, `pvtools — EPICS PV Access command-line tool + +Usage: + pvtools get [pv...] Get current value(s) + pvtools put Write a value + pvtools monitor [pv...] Stream live updates (Ctrl-C to stop) + pvtools info [pv...] Print structure info + +Environment: + EPICS_PVA_ADDR_LIST Space-separated PVA server addresses`) + os.Exit(1) +} + +func main() { + if len(os.Args) < 2 { + usage() + } + + cmd := os.Args[1] + args := os.Args[2:] + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + switch cmd { + case "get": + if len(args) == 0 { + fatalf("get: no PV names specified") + } + runGet(ctx, args) + case "put": + if len(args) < 2 { + fatalf("put: usage: pvtools put ") + } + fatalf("put: PVA write not yet implemented") + case "monitor": + if len(args) == 0 { + fatalf("monitor: no PV names specified") + } + runMonitor(ctx, args) + case "info": + if len(args) == 0 { + fatalf("info: no PV names specified") + } + runInfo(ctx, args) + default: + fatalf("unknown command %q — use get, put, monitor, or info", cmd) + } +} + +// -------------------------------------------------------------------------- // +// get // +// -------------------------------------------------------------------------- // + +func runGet(ctx context.Context, pvs []string) { + cl, err := pva.NewClient() + if err != nil { + fatalf("connect: %v", err) + } + defer cl.Close() + + tctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + for _, name := range pvs { + sv, err := cl.Get(tctx, name) + if err != nil { + fmt.Fprintf(os.Stderr, "%s: error: %v\n", name, err) + continue + } + fmt.Printf("%-30s %s\n", name, formatValue(sv)) + } +} + +// -------------------------------------------------------------------------- // +// monitor // +// -------------------------------------------------------------------------- // + +func runMonitor(ctx context.Context, pvs []string) { + cl, err := pva.NewClient() + if err != nil { + fatalf("connect: %v", err) + } + defer cl.Close() + + for _, name := range pvs { + ch := cl.Monitor(ctx, name) + go func(pvName string, ch <-chan pva.MonitorEvent) { + for evt := range ch { + if evt.Err != nil { + fmt.Fprintf(os.Stderr, "%s: %v\n", pvName, evt.Err) + return + } + ts := time.Now().Local().Format("15:04:05.000") + fmt.Printf("%s %-30s %s\n", ts, pvName, formatValue(evt.Value)) + } + }(name, ch) + fmt.Printf("Monitoring %s (PVA) — press Ctrl-C to stop\n", name) + } + <-ctx.Done() +} + +// -------------------------------------------------------------------------- // +// info // +// -------------------------------------------------------------------------- // + +func runInfo(ctx context.Context, pvs []string) { + cl, err := pva.NewClient() + if err != nil { + fatalf("connect: %v", err) + } + defer cl.Close() + + tctx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + + for _, name := range pvs { + sv, err := cl.Get(tctx, name) + if err != nil { + fmt.Fprintf(os.Stderr, "%s: error: %v\n", name, err) + continue + } + printInfo(name, sv) + fmt.Println() + } +} + +func printInfo(name string, sv pvdata.StructValue) { + fmt.Printf("Name : %s\n", name) + fmt.Printf("Fields : %d\n", len(sv.Fields)) + printStruct(sv, " ") +} + +func printStruct(sv pvdata.StructValue, indent string) { + for _, f := range sv.Fields { + switch v := f.Value.(type) { + case pvdata.StructValue: + fmt.Printf("%s%s (struct):\n", indent, f.Name) + printStruct(v, indent+" ") + default: + fmt.Printf("%s%s = %v\n", indent, f.Name, v) + } + } +} + +// -------------------------------------------------------------------------- // +// Formatting helpers // +// -------------------------------------------------------------------------- // + +func formatValue(sv pvdata.StructValue) string { + if len(sv.Fields) == 0 { + return "" + } + val := fieldByName(sv, "value") + alarm := "" + if a := structByName(sv, "alarm"); a != nil { + if sev := fieldByName(*a, "severity"); sev != nil { + if s, ok := sev.(int32); ok && s != 0 { + alarm = fmt.Sprintf(" (alarm sev=%d)", s) + } + } + } + ts := "" + if t := structByName(sv, "timeStamp"); t != nil { + sec := int64(0) + nsec := int32(0) + if v := fieldByName(*t, "secondsPastEpoch"); v != nil { + sec, _ = v.(int64) + } + if v := fieldByName(*t, "nanoseconds"); v != nil { + nsec, _ = v.(int32) + } + if sec != 0 { + const epicsEpoch = 631152000 + ut := time.Unix(sec+epicsEpoch, int64(nsec)).Local() + ts = " [" + ut.Format("2006-01-02 15:04:05.000") + "]" + } + } + if val == nil { + val = sv.Fields[0].Value + } + return fmt.Sprintf("%v%s%s", formatScalar(val), alarm, ts) +} + +func formatScalar(v any) string { + switch x := v.(type) { + case float64: + return strconv.FormatFloat(x, 'g', -1, 64) + case float32: + return strconv.FormatFloat(float64(x), 'g', -1, 32) + case string: + return x + case []string: + return strings.Join(x, " ") + default: + return fmt.Sprintf("%v", v) + } +} + +func fieldByName(sv pvdata.StructValue, name string) interface{} { + for _, f := range sv.Fields { + if f.Name == name { + return f.Value + } + } + return nil +} + +func structByName(sv pvdata.StructValue, name string) *pvdata.StructValue { + for _, f := range sv.Fields { + if f.Name == name { + if s, ok := f.Value.(pvdata.StructValue); ok { + return &s + } + } + } + return nil +} + +func fatalf(format string, args ...any) { + fmt.Fprintf(os.Stderr, "pvtools: "+format+"\n", args...) + os.Exit(1) +} diff --git a/cmd/uopi/main.go b/cmd/uopi/main.go index b4a2671..5232c04 100644 --- a/cmd/uopi/main.go +++ b/cmd/uopi/main.go @@ -14,6 +14,7 @@ import ( "github.com/uopi/uopi/internal/broker" "github.com/uopi/uopi/internal/config" "github.com/uopi/uopi/internal/datasource/epics" + "github.com/uopi/uopi/internal/datasource/pva" "github.com/uopi/uopi/internal/datasource/stub" "github.com/uopi/uopi/internal/datasource/synthetic" "github.com/uopi/uopi/internal/server" @@ -95,6 +96,16 @@ func main() { } } + // PV Access data source. + if cfg.Datasource.PVA.Enabled { + pvaDS := pva.New(cfg.Datasource.PVA.AddrList) + if err := pvaDS.Connect(ctx); err != nil { + log.Error("pva connect", "err", err) + } else { + brk.Register(pvaDS) + } + } + srv := server.New(cfg.Server.Listen, webFS, brk, synthDS, store, cfg.Datasource.EPICS.ChannelFinderURL, log) if err := srv.Start(ctx); err != nil { diff --git a/go.work b/go.work index 21738f5..ab75d75 100644 --- a/go.work +++ b/go.work @@ -3,4 +3,5 @@ go 1.26.2 use ( . ./pkg/ca + ./pkg/pva ) diff --git a/internal/config/config.go b/internal/config/config.go index e5b1c92..b589ff2 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -21,6 +21,7 @@ type ServerConfig struct { type DatasourceConfig struct { Stub StubConfig `toml:"stub"` EPICS EPICSConfig `toml:"epics"` + PVA PVAConfig `toml:"pva"` Synthetic SyntheticConfig `toml:"synthetic"` } @@ -36,6 +37,11 @@ type EPICSConfig struct { PVNames []string `toml:"pv_names"` } +type PVAConfig struct { + Enabled bool `toml:"enabled"` + AddrList []string `toml:"addr_list"` +} + type SyntheticConfig struct { Enabled bool `toml:"enabled"` } @@ -49,6 +55,7 @@ func Default() Config { Datasource: DatasourceConfig{ Stub: StubConfig{Enabled: true}, EPICS: EPICSConfig{Enabled: true}, + PVA: PVAConfig{Enabled: true}, Synthetic: SyntheticConfig{Enabled: true}, }, } @@ -86,6 +93,9 @@ func applyEnv(cfg *Config) { if v := env("UOPI_EPICS_CHANNEL_FINDER_URL"); v != "" { cfg.Datasource.EPICS.ChannelFinderURL = v } + if v := env("EPICS_PVA_ADDR_LIST"); v != "" { + cfg.Datasource.PVA.AddrList = strings.Fields(v) + } } func env(key string) string { diff --git a/internal/datasource/epics/noop.go b/internal/datasource/epics/noop.go index 902c008..9caef84 100644 --- a/internal/datasource/epics/noop.go +++ b/internal/datasource/epics/noop.go @@ -169,19 +169,15 @@ func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasour func (e *EPICS) timeValueToDS(signal string, tv proto.TimeValue) datasource.Value { var data any if tv.Doubles != nil { - data = tv.Doubles + if len(tv.Doubles) == 1 { + data = tv.Doubles[0] + } else { + data = tv.Doubles + } } else if tv.Str != "" { data = tv.Str } else { - // Use cached metadata type for correct int64 / float64 distinction. - e.mu.RLock() - meta, hasMeta := e.metadata[signal] - e.mu.RUnlock() - if hasMeta && (meta.Type == datasource.TypeInt64 || meta.Type == datasource.TypeEnum) { - data = int64(tv.Long) - } else { - data = tv.Double - } + data = tv.Double } quality := datasource.QualityGood diff --git a/internal/datasource/pva/pva.go b/internal/datasource/pva/pva.go new file mode 100644 index 0000000..fddaf18 --- /dev/null +++ b/internal/datasource/pva/pva.go @@ -0,0 +1,365 @@ +// Package pva provides an EPICS PV Access data source for uopi using the +// pure-Go gopva library. +package pva + +import ( + "context" + "fmt" + "log/slog" + "os" + "strings" + "sync" + "time" + + gopva "github.com/uopi/gopva" + "github.com/uopi/gopva/pvdata" + "github.com/uopi/uopi/internal/datasource" +) + +// PVA is the PV Access data source. +type PVA struct { + client *gopva.Client + + mu sync.RWMutex + metadata map[string]datasource.Metadata +} + +// New creates a new PV Access data source. +// Server addresses are read from the EPICS_PVA_ADDR_LIST environment variable; +// addrList overrides that when non-empty. +func New(addrList []string) datasource.DataSource { + if len(addrList) > 0 { + // Inject addresses into the environment so gopva.NewClient picks them up. + _ = os.Setenv("EPICS_PVA_ADDR_LIST", strings.Join(addrList, " ")) + } + return &PVA{ + metadata: make(map[string]datasource.Metadata), + } +} + +// Name implements datasource.DataSource. +func (p *PVA) Name() string { return "pva" } + +// Connect creates the PVA client. +func (p *PVA) Connect(_ context.Context) error { + cl, err := gopva.NewClient() + if err != nil { + return fmt.Errorf("pva: connect: %w", err) + } + p.client = cl + slog.Info("pva: client started") + return nil +} + +// -------------------------------------------------------------------------- // +// Subscribe // +// -------------------------------------------------------------------------- // + +// Subscribe registers ch to receive live PVA value updates for signal. +func (p *PVA) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) { + slog.Info("pva: subscribing", "pv", signal) + + monCh := p.client.Monitor(ctx, signal) + + // Wait for the first event to confirm the subscription is live. + // We do this in the goroutine to avoid blocking the caller. + done := make(chan struct{}) + go func() { + defer close(done) + for evt := range monCh { + if evt.Err != nil { + slog.Error("pva: monitor error", "pv", signal, "err", evt.Err) + select { + case ch <- datasource.Value{ + Timestamp: time.Now(), + Data: nil, + Quality: datasource.QualityBad, + }: + default: + } + return + } + // Cache metadata from the first value. + p.updateMetadata(signal, evt.Value) + + val := structToValue(evt.Value) + select { + case ch <- val: + default: + } + } + }() + + cancel := func() { + // cancelling the context passed to Monitor stops it; the channel will be closed. + // The goroutine will exit when monCh is closed. + <-done + } + return datasource.CancelFunc(cancel), nil +} + +// -------------------------------------------------------------------------- // +// GetMetadata // +// -------------------------------------------------------------------------- // + +// GetMetadata returns metadata for signal. It performs a one-shot GET if +// metadata has not been cached yet. +func (p *PVA) GetMetadata(ctx context.Context, signal string) (datasource.Metadata, error) { + p.mu.RLock() + if m, ok := p.metadata[signal]; ok { + p.mu.RUnlock() + return m, nil + } + p.mu.RUnlock() + + tctx, cancel := context.WithTimeout(ctx, 15*time.Second) + defer cancel() + + sv, err := p.client.Get(tctx, signal) + if err != nil { + return datasource.Metadata{}, fmt.Errorf("pva: GetMetadata %q: %w", signal, err) + } + + p.updateMetadata(signal, sv) + + p.mu.RLock() + m := p.metadata[signal] + p.mu.RUnlock() + return m, nil +} + +// updateMetadata derives and caches metadata from a StructValue. +func (p *PVA) updateMetadata(signal string, sv pvdata.StructValue) { + m := structToMeta(signal, sv) + p.mu.Lock() + p.metadata[signal] = m + p.mu.Unlock() +} + +// -------------------------------------------------------------------------- // +// ListSignals // +// -------------------------------------------------------------------------- // + +// ListSignals returns metadata for all signals with cached information. +func (p *PVA) ListSignals(_ context.Context) ([]datasource.Metadata, error) { + p.mu.RLock() + defer p.mu.RUnlock() + out := make([]datasource.Metadata, 0, len(p.metadata)) + for _, m := range p.metadata { + out = append(out, m) + } + return out, nil +} + +// -------------------------------------------------------------------------- // +// Write // +// -------------------------------------------------------------------------- // + +// Write is not yet implemented for PVA. +func (p *PVA) Write(_ context.Context, _ string, _ any) error { + return fmt.Errorf("pva: write not yet implemented") +} + +// -------------------------------------------------------------------------- // +// History // +// -------------------------------------------------------------------------- // + +// History is not supported for PVA sources. +func (p *PVA) History(_ context.Context, _ string, _, _ time.Time, _ int) ([]datasource.Value, error) { + return nil, datasource.ErrHistoryUnavailable +} + +// -------------------------------------------------------------------------- // +// Helpers // +// -------------------------------------------------------------------------- // + +// structToValue converts a PVA StructValue to a datasource.Value. +// It follows NTScalar/NTScalarArray conventions: looks for a "value" field +// and an optional "alarm" struct for quality, and "timeStamp" for the timestamp. +func structToValue(sv pvdata.StructValue) datasource.Value { + val := fieldByName(sv, "value") + ts := extractTimestamp(sv) + quality := extractQuality(sv) + + var data any + if val != nil { + data = normaliseValue(val) + } + + return datasource.Value{ + Timestamp: ts, + Data: data, + Quality: quality, + } +} + +// structToMeta derives datasource.Metadata from a PVA StructValue. +func structToMeta(name string, sv pvdata.StructValue) datasource.Metadata { + m := datasource.Metadata{ + Name: name, + Writable: true, // PVA channels are assumed writable until proven otherwise + } + + val := fieldByName(sv, "value") + if val == nil && len(sv.Fields) > 0 { + val = sv.Fields[0].Value + } + + if val != nil { + switch val.(type) { + case float64, float32: + m.Type = datasource.TypeFloat64 + case int64, int32, int16, int8, uint64, uint32, uint16, uint8: + m.Type = datasource.TypeInt64 + case string: + m.Type = datasource.TypeString + case bool: + m.Type = datasource.TypeBool + case []float64, []float32: + m.Type = datasource.TypeFloat64Array + default: + m.Type = datasource.TypeFloat64 + } + } + + // Extract display limits from NTScalar display structure. + if disp := structByName(sv, "display"); disp != nil { + if v := fieldByName(*disp, "limitLow"); v != nil { + m.DisplayLow, _ = toFloat64(v) + } + if v := fieldByName(*disp, "limitHigh"); v != nil { + m.DisplayHigh, _ = toFloat64(v) + } + if v := fieldByName(*disp, "units"); v != nil { + m.Unit, _ = v.(string) + } + } + + // Extract control limits from NTScalar control structure. + if ctrl := structByName(sv, "control"); ctrl != nil { + if v := fieldByName(*ctrl, "limitLow"); v != nil { + m.DriveLow, _ = toFloat64(v) + } + if v := fieldByName(*ctrl, "limitHigh"); v != nil { + m.DriveHigh, _ = toFloat64(v) + } + } + + return m +} + +// normaliseValue converts any PVA scalar/array value to datasource-compatible types. +func normaliseValue(v any) any { + switch x := v.(type) { + case float64: + return x + case float32: + return float64(x) + case int64: + return x + case int32: + return int64(x) + case int16: + return int64(x) + case int8: + return int64(x) + case uint64: + return int64(x) + case uint32: + return int64(x) + case uint16: + return int64(x) + case uint8: + return int64(x) + case bool: + return x + case string: + return x + case []float64: + return x + case []float32: + out := make([]float64, len(x)) + for i, f := range x { + out[i] = float64(f) + } + return out + default: + return fmt.Sprintf("%v", v) + } +} + +func extractTimestamp(sv pvdata.StructValue) time.Time { + ts := structByName(sv, "timeStamp") + if ts == nil { + return time.Now() + } + sec := int64(0) + nsec := int32(0) + if v := fieldByName(*ts, "secondsPastEpoch"); v != nil { + sec, _ = v.(int64) + } + if v := fieldByName(*ts, "nanoseconds"); v != nil { + nsec, _ = v.(int32) + } + if sec == 0 { + return time.Now() + } + // EPICS epoch: 1990-01-01 00:00:00 UTC = Unix 631152000 + const epicsEpoch = 631152000 + return time.Unix(sec+epicsEpoch, int64(nsec)).UTC() +} + +func extractQuality(sv pvdata.StructValue) datasource.Quality { + alarm := structByName(sv, "alarm") + if alarm == nil { + return datasource.QualityGood + } + sev := fieldByName(*alarm, "severity") + if sev == nil { + return datasource.QualityGood + } + s, _ := sev.(int32) + switch s { + case 1: + return datasource.QualityUncertain + case 2, 3: + return datasource.QualityBad + default: + return datasource.QualityGood + } +} + +func toFloat64(v any) (float64, bool) { + switch x := v.(type) { + case float64: + return x, true + case float32: + return float64(x), true + case int64: + return float64(x), true + case int32: + return float64(x), true + default: + return 0, false + } +} + +func fieldByName(sv pvdata.StructValue, name string) any { + for _, f := range sv.Fields { + if f.Name == name { + return f.Value + } + } + return nil +} + +func structByName(sv pvdata.StructValue, name string) *pvdata.StructValue { + for _, f := range sv.Fields { + if f.Name == name { + if s, ok := f.Value.(pvdata.StructValue); ok { + return &s + } + } + } + return nil +} diff --git a/pkg/ca/conn.go b/pkg/ca/conn.go index fedc141..869f3ae 100644 --- a/pkg/ca/conn.go +++ b/pkg/ca/conn.go @@ -53,19 +53,17 @@ type monState struct { // - sid, dbfType, count, access: valid only while readyC is closed. // - readyC: closed when CREATE_CHAN reply received; replaced on reconnect. // - monitors: append-only (except removeMonitor); never cleared on reconnect. -// - pending: ioid → GET reply channel; drained (channels closed) on reconnect. type chanState struct { cid uint32 pvName string mu sync.RWMutex - sid uint32 // server-assigned channel ID (0 until CREATE_CHAN reply) - dbfType int // native DBF field type - count uint32 // element count - access uint32 // AccessRead | AccessWrite bitmask - readyC chan struct{} // closed once CREATE_CHAN reply is received - monitors []*monState // active subscriptions - pending map[uint32]chan reply // ioid → one-shot GET callback + sid uint32 // server-assigned channel ID (0 until CREATE_CHAN reply) + dbfType int // native DBF field type + count uint32 // element count + access uint32 // AccessRead | AccessWrite bitmask + readyC chan struct{} // closed once CREATE_CHAN reply is received + monitors []*monState // active subscriptions } func newChanState(cid uint32, pvName string) *chanState { @@ -73,7 +71,6 @@ func newChanState(cid uint32, pvName string) *chanState { cid: cid, pvName: pvName, readyC: make(chan struct{}), - pending: make(map[uint32]chan reply), } } @@ -81,16 +78,12 @@ func newChanState(cid uint32, pvName string) *chanState { // It closes the current readyC (waking any waitReady callers so they can // re-wait on the fresh channel) and then replaces it. // Must be called before the circuit sends CREATE_CHAN on the new conn. +// In-flight GET requests are tracked in circuit.byIOID and cleared there. func (cs *chanState) resetForReconnect() { cs.mu.Lock() cs.sid = 0 old := cs.readyC cs.readyC = make(chan struct{}) - // Drain pending GETs — they will receive zero reply (ok=false). - for id, ch := range cs.pending { - close(ch) - delete(cs.pending, id) - } cs.mu.Unlock() // Close the old readyC outside the lock to avoid deadlock with waitReady. select { @@ -150,7 +143,9 @@ type circuit struct { mu sync.RWMutex channels []*chanState // append-only; survive reconnect byCID map[uint32]*chanState // cid → chanState (fast lookup) + bySID map[uint32]*chanState // sid → chanState (fast lookup) bySubID map[uint32]*monState // subID → monState (fast event dispatch) + byIOID map[uint32]chan reply // ioid → GET/PUT callback (circuit-wide) writeQ chan []byte // serialised outbound message queue seq atomic.Uint32 // shared ID sequence (cid, ioid, subID) @@ -165,7 +160,9 @@ func newCircuit(ctx context.Context, addr, clientName, hostName string) *circuit ctx: cctx, cancel: cancel, byCID: make(map[uint32]*chanState), + bySID: make(map[uint32]*chanState), bySubID: make(map[uint32]*monState), + byIOID: make(map[uint32]chan reply), writeQ: make(chan []byte, writeQueueDepth), } go c.run() @@ -204,6 +201,12 @@ func (c *circuit) run() { // Snapshot channels and reset their per-connection state. c.mu.Lock() + c.bySID = make(map[uint32]*chanState) + // Close all in-flight GET reply channels so waiters unblock with ok=false. + for _, ch := range c.byIOID { + close(ch) + } + c.byIOID = make(map[uint32]chan reply) for _, cs := range c.channels { cs.resetForReconnect() } @@ -361,9 +364,13 @@ func (c *circuit) dispatch(conn net.Conn, hdr proto.Header, payload []byte) { case proto.CmdCreateChan: // Parameter1 = cid (echoed), Parameter2 = SID assigned by server. - c.mu.RLock() + c.mu.Lock() cs, ok := c.byCID[hdr.Parameter1] - c.mu.RUnlock() + if ok { + c.bySID[hdr.Parameter2] = cs + } + c.mu.Unlock() + if !ok { return } @@ -425,21 +432,23 @@ func (c *circuit) dispatch(conn net.Conn, hdr proto.Header, payload []byte) { } case proto.CmdEventAdd: - // Monitor update: Parameter1 = subscriptionID. + // Monitor update: server puts subscription ID in Parameter2 (m_available). + // Parameter1 carries ECA_NORMAL (1) as a status code. + subID := hdr.Parameter2 c.mu.RLock() - ms, ok := c.bySubID[hdr.Parameter1] + ms, ok := c.bySubID[subID] c.mu.RUnlock() if !ok { - dbg("CA EVENT_ADD for unknown subID", "subID", hdr.Parameter1) + dbg("CA EVENT_ADD for unknown subID", "subID", subID) return } tv, ok := proto.DecodeTimeValue(hdr.DataType, hdr.DataCount, payload) if !ok { - dbg("CA EVENT_ADD decode failed", "subID", hdr.Parameter1, + dbg("CA EVENT_ADD decode failed", "subID", subID, "dbrType", hdr.DataType, "count", hdr.DataCount, "payloadLen", len(payload)) return } - dbg("CA EVENT_ADD value", "subID", hdr.Parameter1, "dbrType", hdr.DataType, + dbg("CA EVENT_ADD value", "subID", subID, "dbrType", hdr.DataType, "double", tv.Double, "severity", tv.Severity) select { case ms.ch <- tv: @@ -447,25 +456,20 @@ func (c *circuit) dispatch(conn net.Conn, hdr proto.Header, payload []byte) { } case proto.CmdReadNotify: - // GET reply: Parameter1 = ioid. - ioid := hdr.Parameter1 - c.mu.RLock() - var replyCh chan reply - var found bool - for _, cs := range c.channels { - cs.mu.Lock() - if ch, exists := cs.pending[ioid]; exists { - delete(cs.pending, ioid) - replyCh = ch - found = true - } - cs.mu.Unlock() - if found { - break - } + // Parameter2 = ioid (our request ID echoed back by server). + ioid := hdr.Parameter2 + + c.mu.Lock() + replyCh, ok := c.byIOID[ioid] + if ok { + delete(c.byIOID, ioid) } - c.mu.RUnlock() - if found { + c.mu.Unlock() + + dbg("CA CmdReadNotify", "ioid", ioid, "found", ok, "dbrType", hdr.DataType, + "count", hdr.DataCount, "payloadSize", hdr.PayloadSize) + + if ok { select { case replyCh <- reply{hdr: hdr, payload: payload, ok: true}: default: @@ -615,9 +619,9 @@ func (c *circuit) getRaw(ctx context.Context, cs *chanState, dbrType uint16, cou ioid := c.nextID() replyCh := make(chan reply, 1) - cs.mu.Lock() - cs.pending[ioid] = replyCh - cs.mu.Unlock() + c.mu.Lock() + c.byIOID[ioid] = replyCh + c.mu.Unlock() msg := proto.BuildMessage(proto.Header{ Command: proto.CmdReadNotify, @@ -630,9 +634,9 @@ func (c *circuit) getRaw(ctx context.Context, cs *chanState, dbrType uint16, cou select { case c.writeQ <- msg: case <-ctx.Done(): - cs.mu.Lock() - delete(cs.pending, ioid) - cs.mu.Unlock() + c.mu.Lock() + delete(c.byIOID, ioid) + c.mu.Unlock() return proto.Header{}, nil, ctx.Err() } @@ -643,9 +647,9 @@ func (c *circuit) getRaw(ctx context.Context, cs *chanState, dbrType uint16, cou } return r.hdr, r.payload, nil case <-ctx.Done(): - cs.mu.Lock() - delete(cs.pending, ioid) - cs.mu.Unlock() + c.mu.Lock() + delete(c.byIOID, ioid) + c.mu.Unlock() return proto.Header{}, nil, ctx.Err() } } @@ -667,9 +671,9 @@ func (c *circuit) get(ctx context.Context, cs *chanState, dbrType uint16, count ioid := c.nextID() replyCh := make(chan reply, 1) - cs.mu.Lock() - cs.pending[ioid] = replyCh - cs.mu.Unlock() + c.mu.Lock() + c.byIOID[ioid] = replyCh + c.mu.Unlock() msg := proto.BuildMessage(proto.Header{ Command: proto.CmdReadNotify, @@ -682,9 +686,9 @@ func (c *circuit) get(ctx context.Context, cs *chanState, dbrType uint16, count select { case c.writeQ <- msg: case <-ctx.Done(): - cs.mu.Lock() - delete(cs.pending, ioid) - cs.mu.Unlock() + c.mu.Lock() + delete(c.byIOID, ioid) + c.mu.Unlock() return proto.TimeValue{}, ctx.Err() } @@ -699,9 +703,9 @@ func (c *circuit) get(ctx context.Context, cs *chanState, dbrType uint16, count } return tv, nil case <-ctx.Done(): - cs.mu.Lock() - delete(cs.pending, ioid) - cs.mu.Unlock() + c.mu.Lock() + delete(c.byIOID, ioid) + c.mu.Unlock() return proto.TimeValue{}, ctx.Err() } } diff --git a/pkg/ca/proto/dbr.go b/pkg/ca/proto/dbr.go index d639f49..5dba4f2 100644 --- a/pkg/ca/proto/dbr.go +++ b/pkg/ca/proto/dbr.go @@ -85,7 +85,7 @@ func caString(b []byte) string { return string(b[:idx]) } -// DBR_TIME_* wire layouts (all big-endian): +// DBR_TIME_* wire layouts (all big-endian, matching EPICS db_access.h structs): // // Common header (12 bytes): // [0:2] int16 status @@ -93,13 +93,13 @@ func caString(b []byte) string { // [4:8] uint32 secPastEpoch // [8:12] uint32 nsec // -// DBR_TIME_DOUBLE (type 25): 4-byte RISC pad at [12:16], float64 at [16:24]. Total=24. -// DBR_TIME_FLOAT (type 20): float32 at [12:16]. Total=16. -// DBR_TIME_LONG (type 22): int32 at [12:16]. Total=16. -// DBR_TIME_SHORT (type 19): int16 at [12:14], 2-byte pad [14:16]. Total=16. -// DBR_TIME_ENUM (type 23): uint16 at [12:14], 2-byte pad [14:16]. Total=16. -// DBR_TIME_CHAR (type 24): uint8 at [12:13], 3-byte pad [13:16]. Total=16. -// DBR_TIME_STRING (type 21): [40]byte at [12:52]. Total=52. +// DBR_TIME_STRING (type 14): [40]byte at [12:52]. Total=52. +// DBR_TIME_SHORT (type 15): 2-byte RISC pad at [12:14], int16 at [14:16]. Total=16. +// DBR_TIME_FLOAT (type 16): float32 at [12:16]. Total=16. +// DBR_TIME_ENUM (type 17): 2-byte RISC pad at [12:14], uint16 at [14:16]. Total=16. +// DBR_TIME_CHAR (type 18): RISC_pad0[12:14], RISC_pad1[14], uint8 at [15]. Total=16. +// DBR_TIME_LONG (type 19): int32 at [12:16]. Total=16. +// DBR_TIME_DOUBLE (type 20): 4-byte RISC pad at [12:16], float64 at [16:24]. Total=24. // DecodeTimeValue decodes a DBR_TIME_* payload. // dbrType is one of the DBRTime* constants; payload is the full message payload @@ -158,24 +158,27 @@ func DecodeTimeValue(dbrType uint16, count uint32, payload []byte) (TimeValue, b tv.Double = float64(tv.Long) case DBRTimeShort: + // 2-byte RISC pad at [12:14], value at [14:16] if len(payload) < 16 { return TimeValue{}, false } - tv.Short = int16(binary.BigEndian.Uint16(payload[12:])) + tv.Short = int16(binary.BigEndian.Uint16(payload[14:])) tv.Double = float64(tv.Short) case DBRTimeEnum: + // 2-byte RISC pad at [12:14], value at [14:16] if len(payload) < 16 { return TimeValue{}, false } - tv.Enum = binary.BigEndian.Uint16(payload[12:]) + tv.Enum = binary.BigEndian.Uint16(payload[14:]) tv.Double = float64(tv.Enum) case DBRTimeChar: + // RISC_pad0[12:14], RISC_pad1[14], value[15] per db_access.h dbr_time_char if len(payload) < 16 { return TimeValue{}, false } - tv.Char = payload[12] + tv.Char = payload[15] tv.Double = float64(tv.Char) case DBRTimeString: @@ -427,15 +430,15 @@ func EncodeString(v string) []byte { // EncodeEventMask builds the 16-byte payload for a CA_PROTO_EVENT_ADD request. // mask is typically DBEDefault (DBEValue | DBEAlarm). // -// Wire layout (16 bytes, all big-endian): +// Wire layout matches caProto.h struct mon_info (all big-endian): // -// [0:4] float32 m_lval (not used, zero) -// [4:8] float32 p_delta (delta trigger, zero = disabled) -// [8:12] float32 p_final (final trigger, zero = disabled) -// [12:14] int16 p_count (element count, zero = use PV's count) -// [14:16] int16 m_mask (DBE_VALUE | DBE_ALARM | ...) +// [0:4] float32 m_lval (low delta, zero = disabled) +// [4:8] float32 m_hval (high delta, zero = disabled) +// [8:12] float32 m_toval (period between samples, zero = disabled) +// [12:14] uint16 m_mask (DBE_VALUE | DBE_ALARM | ...) +// [14:16] uint16 m_pad (alignment padding) func EncodeEventMask(mask uint16) []byte { b := make([]byte, 16) - binary.BigEndian.PutUint16(b[14:], mask) // m_mask is at offset 14, not 12 + binary.BigEndian.PutUint16(b[12:], mask) // m_mask is at offset 12 per caProto.h return b } diff --git a/pkg/ca/proto/opcodes.go b/pkg/ca/proto/opcodes.go index 62d6327..58ec870 100644 --- a/pkg/ca/proto/opcodes.go +++ b/pkg/ca/proto/opcodes.go @@ -74,14 +74,15 @@ const ( ) // DBR_TIME_* types (value + timestamp + alarm status; used in EVENT_ADD). +// Numbers from db_access.h: STRING=14, SHORT/INT=15, FLOAT=16, ENUM=17, CHAR=18, LONG=19, DOUBLE=20. const ( - DBRTimeString = 21 - DBRTimeShort = 19 - DBRTimeFloat = 20 - DBRTimeEnum = 23 - DBRTimeChar = 24 - DBRTimeLong = 22 - DBRTimeDouble = 25 + DBRTimeString = 14 + DBRTimeShort = 15 + DBRTimeFloat = 16 + DBRTimeEnum = 17 + DBRTimeChar = 18 + DBRTimeLong = 19 + DBRTimeDouble = 20 ) // DBR_CTRL_* types (full control info: units, limits, enum strings; used in READ_NOTIFY). diff --git a/pkg/ca/proto/proto_test.go b/pkg/ca/proto/proto_test.go index 17360ae..fe205d9 100644 --- a/pkg/ca/proto/proto_test.go +++ b/pkg/ca/proto/proto_test.go @@ -159,7 +159,7 @@ func TestDecodeTimeString(t *testing.T) { func TestDecodeTimeEnum(t *testing.T) { hdr := buildTimeHeader(0, 0, 0, 0) - val := []byte{0x00, 0x03, 0x00, 0x00} // enum=3 + 2-byte pad + val := []byte{0x00, 0x00, 0x00, 0x03} // 2-byte RISC pad + enum=3 payload := append(hdr, val...) tv, ok := proto.DecodeTimeValue(proto.DBRTimeEnum, 1, payload) @@ -307,14 +307,14 @@ func TestEncodeEventMask(t *testing.T) { if len(b) != 16 { t.Fatalf("len = %d, want 16", len(b)) } - // m_mask is at offset 14 (after m_lval[0:4], p_delta[4:8], p_final[8:12], p_count[12:14]). - mask := binary.BigEndian.Uint16(b[14:]) + // m_mask is at offset 12 per caProto.h struct mon_info. + mask := binary.BigEndian.Uint16(b[12:]) if mask != proto.DBEDefault { t.Errorf("mask = 0x%02X, want 0x%02X", mask, proto.DBEDefault) } - // p_count at [12:14] must be zero. - if pc := binary.BigEndian.Uint16(b[12:]); pc != 0 { - t.Errorf("p_count = %d, want 0", pc) + // m_pad at [14:16] must be zero. + if pad := binary.BigEndian.Uint16(b[14:]); pad != 0 { + t.Errorf("m_pad = %d, want 0", pad) } } diff --git a/pkg/ca/testca/server.go b/pkg/ca/testca/server.go index d0afe12..5403022 100644 --- a/pkg/ca/testca/server.go +++ b/pkg/ca/testca/server.go @@ -338,9 +338,11 @@ func (s *Server) handleConn(conn net.Conn) { ioid := hdr.Parameter2 var pvName string - for _, info := range channels { + var cid uint32 + for c, info := range channels { if info.sid == sid { pvName = info.pvName + cid = c break } } @@ -368,7 +370,8 @@ func (s *Server) handleConn(conn net.Conn) { Command: proto.CmdReadNotify, DataType: hdr.DataType, DataCount: hdr.DataCount, - Parameter1: ioid, + Parameter1: cid, // echo client channel ID (matches real EPICS IOC) + Parameter2: ioid, // echo ioid so client can match the reply }, replyPayload) conn.Write(reply) @@ -430,7 +433,8 @@ func (s *Server) buildEventMsg(sub *serverSub, pv *serverPV) []byte { Command: proto.CmdEventAdd, DataType: sub.dbrType, DataCount: sub.count, - Parameter1: sub.subID, + Parameter1: 1, // ECA_NORMAL + Parameter2: sub.subID, }, payload) } @@ -461,12 +465,12 @@ func (s *Server) encodeTimeValue(dbrType uint16, _ uint32, val any) []byte { binary.BigEndian.PutUint32(body, uint32(int32(toF64(val)))) case proto.DBRTimeShort, proto.DBRTimeEnum: - body = make([]byte, 4) // 2 value + 2 pad - binary.BigEndian.PutUint16(body, uint16(int16(toF64(val)))) + 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) // 1 value + 3 pad - body[0] = byte(uint8(toF64(val))) + 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) diff --git a/pkg/pva/client.go b/pkg/pva/client.go new file mode 100644 index 0000000..1463527 --- /dev/null +++ b/pkg/pva/client.go @@ -0,0 +1,340 @@ +package pva + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "log/slog" + "net" + "sync" + "sync/atomic" + "time" + + "github.com/uopi/gopva/pvdata" +) + +// DefaultTCPPort is the standard PVA TCP port for channel connections. +const DefaultTCPPort = 5075 + +// DefaultUDPPort is the PVA broadcast/search port (distinct from TCP). +const DefaultUDPPort = 5076 + +// Client is a PVA client that manages connections to one or more servers. +// It performs UDP search to locate PVs, then opens TCP connections on demand. +type Client struct { + mu sync.Mutex + conns map[string]*conn // server addr → connection + pending map[string][]waiter // pvName → waiters for address + + searchSeq atomic.Uint32 + udp *net.UDPConn + + ctx context.Context + cancel context.CancelFunc +} + +type waiter struct { + pvName string + cb func(addr string, err error) +} + +// NewClient creates a PVA client. Call Close when done. +func NewClient() (*Client, error) { + ctx, cancel := context.WithCancel(context.Background()) + cl := &Client{ + conns: make(map[string]*conn), + pending: make(map[string][]waiter), + ctx: ctx, + cancel: cancel, + } + if err := cl.startUDP(); err != nil { + cancel() + return nil, err + } + return cl, nil +} + +// Close shuts down the client and all connections. +func (cl *Client) Close() { + cl.cancel() + if cl.udp != nil { + cl.udp.Close() + } + cl.mu.Lock() + defer cl.mu.Unlock() + for _, c := range cl.conns { + c.close() + } +} + +// Get issues a one-shot GET for pvName and returns the decoded structure. +func (cl *Client) Get(ctx context.Context, pvName string) (pvdata.StructValue, error) { + ch := make(chan struct { + v pvdata.StructValue + err error + }, 1) + cl.withConn(ctx, pvName, func(c *conn, err error) { + if err != nil { + ch <- struct { + v pvdata.StructValue + err error + }{err: err} + return + } + c.get(pvName, func(v pvdata.StructValue, e error) { + ch <- struct { + v pvdata.StructValue + err error + }{v, e} + }) + }) + select { + case res := <-ch: + return res.v, res.err + case <-ctx.Done(): + return pvdata.StructValue{}, ctx.Err() + } +} + +// Monitor subscribes to pvName and returns a channel that receives updates. +// The channel is closed when ctx is cancelled or a fatal error occurs. +func (cl *Client) Monitor(ctx context.Context, pvName string) <-chan MonitorEvent { + ch := make(chan MonitorEvent, 16) + cl.withConn(ctx, pvName, func(c *conn, err error) { + if err != nil { + ch <- MonitorEvent{Err: err} + close(ch) + return + } + c.subscribe(ctx, pvName, ch) + }) + return ch +} + +// ---- Internal -------------------------------------------------------- + +// withConn locates (or creates) a connection to the server hosting pvName +// and calls cb on it. cb may be called from a goroutine. +func (cl *Client) withConn(ctx context.Context, pvName string, cb func(*conn, error)) { + cl.search(ctx, pvName, func(addr string, err error) { + if err != nil { + cb(nil, err) + return + } + cl.mu.Lock() + c, ok := cl.conns[addr] + cl.mu.Unlock() + if ok { + cb(c, nil) + return + } + // open new TCP connection + tc, err := dial(ctx, addr) + if err != nil { + cb(nil, fmt.Errorf("pva: connect %s: %w", addr, err)) + return + } + cl.mu.Lock() + cl.conns[addr] = tc + cl.mu.Unlock() + cb(tc, nil) + }) +} + +// ---- UDP search ------------------------------------------------------- + +func (cl *Client) startUDP() error { + conn, err := net.ListenUDP("udp4", &net.UDPAddr{Port: 0}) + if err != nil { + return fmt.Errorf("pva: UDP listen: %w", err) + } + cl.udp = conn + go cl.udpReadLoop() + return nil +} + +// search sends a PVA search request and calls cb when a response arrives. +// Currently does broadcast + localhost; production code would also check +// EPICS_PVA_ADDR_LIST environment variable. +func (cl *Client) search(ctx context.Context, pvName string, cb func(string, error)) { + seq := cl.searchSeq.Add(1) + + cl.mu.Lock() + cl.pending[pvName] = append(cl.pending[pvName], waiter{pvName: pvName, cb: cb}) + cl.mu.Unlock() + + go func() { + for attempt := 0; attempt < 10; attempt++ { + select { + case <-ctx.Done(): + cl.mu.Lock() + cl.removeWaiter(pvName, cb) + cl.mu.Unlock() + cb("", ctx.Err()) + return + default: + } + cl.sendSearchRequest(seq, pvName) + // exponential backoff: 100ms, 200ms, 400ms … up to 2s + delay := time.Duration(100<= len(bs.bits) { + return false + } + return bs.bits[byteIdx]&(1<<(uint(i)%8)) != 0 +} + +// ReadBitSet reads a BitSet from r. +func ReadBitSet(r io.Reader) (BitSet, error) { + n, err := ReadSize(r) + if err != nil { + return BitSet{}, err + } + if n == 0 { + return BitSet{}, nil + } + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return BitSet{}, err + } + return BitSet{bits: buf}, nil +} + +// WriteBitSet writes bs to w. +func WriteBitSet(w io.Writer, bs BitSet) error { + if err := WriteSize(w, int64(len(bs.bits))); err != nil { + return err + } + if len(bs.bits) == 0 { + return nil + } + _, err := w.Write(bs.bits) + return err +} + +// FieldIndices returns all set bit indices in ascending order. +func (bs *BitSet) FieldIndices() []int { + var out []int + for bi, b := range bs.bits { + for bit := 0; bit < 8; bit++ { + if b&(1<= 0 { + return int64(v), nil + } + // extended 8-byte size + var v8 int64 + if err := binary.Read(r, binary.LittleEndian, &v8); err != nil { + return 0, err + } + return v8, nil +} + +// WriteSize writes n as a compact-format size to w. +func WriteSize(w io.Writer, n int64) error { + if n < 0xFF { + _, err := w.Write([]byte{byte(n)}) + return err + } + if n <= math.MaxInt32 { + if _, err := w.Write([]byte{0xFF}); err != nil { + return err + } + return binary.Write(w, binary.LittleEndian, int32(n)) + } + // rare: >2 GiB array + if _, err := w.Write([]byte{0xFF}); err != nil { + return err + } + if err := binary.Write(w, binary.LittleEndian, int32(-1)); err != nil { + return err + } + return binary.Write(w, binary.LittleEndian, n) +} + +// ReadString reads a pvData string (compact size + UTF-8 bytes). +func ReadString(r io.Reader) (string, error) { + n, err := ReadSize(r) + if err != nil { + return "", err + } + if n == 0 { + return "", nil + } + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return "", err + } + return string(buf), nil +} + +// WriteString writes a pvData string to w. +func WriteString(w io.Writer, s string) error { + if err := WriteSize(w, int64(len(s))); err != nil { + return err + } + _, err := io.WriteString(w, s) + return err +} + +// ---- Field description (introspection) -------------------------------- + +// Kind classifies a FieldDesc at a high level. +type Kind byte + +const ( + KindScalar Kind = iota // single scalar value + KindScalarArr // variable-length array of scalars + KindString // pvData string (treated specially) + KindStringArr + KindStruct // structure + KindStructArr + KindUnion // discriminated union + KindUnionArr + KindVariant // any (variant) +) + +// FieldDesc describes the type of a single field in a pvData structure. +type FieldDesc struct { + Kind Kind + TypeCode byte // scalar/array type code; 0 for struct/union + TypeID string // optional structure/union type ID string + Fields []Field // child fields for struct/union +} + +// Field is a named FieldDesc (one member of a struct or union). +type Field struct { + Name string + Desc FieldDesc +} + +// ---- FieldDesc codec -------------------------------------------------- + +// ReadFieldDesc decodes a FieldDesc from r. +// This follows the pvData introspection encoding (spec §4). +func ReadFieldDesc(r io.Reader) (FieldDesc, error) { + var tc [1]byte + if _, err := io.ReadFull(r, tc[:]); err != nil { + return FieldDesc{}, err + } + return readFieldDescFromTypeCode(r, tc[0]) +} + +func readFieldDescFromTypeCode(r io.Reader, tc byte) (FieldDesc, error) { + switch { + case tc == TypeCodeNull: + return FieldDesc{TypeCode: TypeCodeNull}, nil + + case tc == TypeCodeString: + return FieldDesc{Kind: KindString, TypeCode: tc}, nil + + case tc == TypeCodeStringArr: + return FieldDesc{Kind: KindStringArr, TypeCode: tc}, nil + + case tc == TypeCodeVariant: + return FieldDesc{Kind: KindVariant, TypeCode: tc}, nil + + case tc == TypeCodeStruct || tc == TypeCodeUnion: + kind := KindStruct + if tc == TypeCodeUnion { + kind = KindUnion + } + typeID, err := ReadString(r) + if err != nil { + return FieldDesc{}, err + } + nfields, err := ReadSize(r) + if err != nil { + return FieldDesc{}, err + } + fields := make([]Field, nfields) + for i := range fields { + name, err := ReadString(r) + if err != nil { + return FieldDesc{}, err + } + desc, err := ReadFieldDesc(r) + if err != nil { + return FieldDesc{}, err + } + fields[i] = Field{Name: name, Desc: desc} + } + return FieldDesc{Kind: kind, TypeCode: tc, TypeID: typeID, Fields: fields}, nil + + case tc == TypeCodeStructArr || tc == TypeCodeUnionArr: + kind := KindStructArr + if tc == TypeCodeUnionArr { + kind = KindUnionArr + } + // element descriptor follows + elem, err := ReadFieldDesc(r) + if err != nil { + return FieldDesc{}, err + } + return FieldDesc{Kind: kind, TypeCode: tc, TypeID: elem.TypeID, Fields: elem.Fields}, nil + + // scalar array types: 0x28–0x2F, 0x4A–0x4B, 0x08, 0x68 + case isScalarArrayCode(tc): + return FieldDesc{Kind: KindScalarArr, TypeCode: tc}, nil + + // plain scalars and boolean + default: + return FieldDesc{Kind: KindScalar, TypeCode: tc}, nil + } +} + +func isScalarArrayCode(tc byte) bool { + switch tc { + case TypeCodeBoolArr, + TypeCodeByteArr, TypeCodeShortArr, TypeCodeIntArr, TypeCodeLongArr, + TypeCodeUByteArr, TypeCodeUShortArr, TypeCodeUIntArr, TypeCodeULongArr, + TypeCodeFloatArr, TypeCodeDoubleArr, + TypeCodeStringArr: + return true + } + return false +} + +// WriteFieldDesc encodes fd into w. +func WriteFieldDesc(w io.Writer, fd FieldDesc) error { + if _, err := w.Write([]byte{fd.TypeCode}); err != nil { + return err + } + switch fd.Kind { + case KindStruct, KindUnion: + if err := WriteString(w, fd.TypeID); err != nil { + return err + } + if err := WriteSize(w, int64(len(fd.Fields))); err != nil { + return err + } + for _, f := range fd.Fields { + if err := WriteString(w, f.Name); err != nil { + return err + } + if err := WriteFieldDesc(w, f.Desc); err != nil { + return err + } + } + case KindStructArr, KindUnionArr: + // write element descriptor + elemTC := TypeCodeStruct + if fd.Kind == KindUnionArr { + elemTC = TypeCodeUnion + } + if err := WriteFieldDesc(w, FieldDesc{ + Kind: KindStruct, TypeCode: byte(elemTC), + TypeID: fd.TypeID, Fields: fd.Fields, + }); err != nil { + return err + } + } + return nil +} + +// ---- Value codec ------------------------------------------------------- + +// Value is a decoded pvData value. We use interface{} for flexibility; +// callers type-assert to concrete types listed below. +// +// Scalar types: +// bool, int8, int16, int32, int64, +// uint8, uint16, uint32, uint64, +// float32, float64, string +// +// Array types: []T for each scalar T above; []string. +// Struct: StructValue +// Union: UnionValue +// Variant: VariantValue +// Null: nil +type Value = interface{} + +// StructValue holds the fields of a decoded pvData structure. +type StructValue struct { + TypeID string + Fields []FieldValue +} + +// FieldValue is a named value within a StructValue. +type FieldValue struct { + Name string + Value Value +} + +// UnionValue holds a selected field from a discriminated union. +type UnionValue struct { + Selected int // index into union's Fields slice + Name string // field name + Value Value +} + +// VariantValue wraps an any-typed value together with its runtime descriptor. +type VariantValue struct { + Desc FieldDesc + Value Value +} + +// ReadValue decodes a value conforming to desc from r. +func ReadValue(r io.Reader, desc FieldDesc) (Value, error) { + switch desc.Kind { + case KindScalar: + return readScalar(r, desc.TypeCode) + case KindScalarArr: + return readScalarArray(r, desc.TypeCode) + case KindString: + return ReadString(r) + case KindStringArr: + n, err := ReadSize(r) + if err != nil { + return nil, err + } + ss := make([]string, n) + for i := range ss { + ss[i], err = ReadString(r) + if err != nil { + return nil, err + } + } + return ss, nil + case KindStruct: + return readStruct(r, desc) + case KindStructArr: + return readStructArr(r, desc) + case KindUnion: + return readUnion(r, desc) + case KindUnionArr: + return readUnionArr(r, desc) + case KindVariant: + return readVariant(r) + default: + return nil, fmt.Errorf("pvdata: unknown kind %d", desc.Kind) + } +} + +func readScalar(r io.Reader, tc byte) (Value, error) { + switch tc { + case TypeCodeBoolean: + var b [1]byte + _, err := io.ReadFull(r, b[:]) + return b[0] != 0, err + case TypeCodeByte: + var v int8 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeShort: + var v int16 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeInt: + var v int32 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeLong: + var v int64 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeUByte: + var v uint8 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeUShort: + var v uint16 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeUInt: + var v uint32 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeULong: + var v uint64 + return v, binary.Read(r, binary.LittleEndian, &v) + case TypeCodeFloat: + var v uint32 + if err := binary.Read(r, binary.LittleEndian, &v); err != nil { + return nil, err + } + return math.Float32frombits(v), nil + case TypeCodeDouble: + var v uint64 + if err := binary.Read(r, binary.LittleEndian, &v); err != nil { + return nil, err + } + return math.Float64frombits(v), nil + default: + return nil, fmt.Errorf("pvdata: unknown scalar type code 0x%02X", tc) + } +} + +func readScalarArray(r io.Reader, tc byte) (Value, error) { + n, err := ReadSize(r) + if err != nil { + return nil, err + } + count := int(n) + switch tc { + case TypeCodeBoolArr: + v := make([]bool, count) + for i := range v { + b := [1]byte{} + if _, err := io.ReadFull(r, b[:]); err != nil { + return nil, err + } + v[i] = b[0] != 0 + } + return v, nil + case TypeCodeByteArr: + v := make([]int8, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeShortArr: + v := make([]int16, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeIntArr: + v := make([]int32, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeLongArr: + v := make([]int64, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeUByteArr: + v := make([]uint8, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeUShortArr: + v := make([]uint16, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeUIntArr: + v := make([]uint32, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeULongArr: + v := make([]uint64, count) + return v, binary.Read(r, binary.LittleEndian, v) + case TypeCodeFloatArr: + bits := make([]uint32, count) + if err := binary.Read(r, binary.LittleEndian, bits); err != nil { + return nil, err + } + v := make([]float32, count) + for i, b := range bits { + v[i] = math.Float32frombits(b) + } + return v, nil + case TypeCodeDoubleArr: + bits := make([]uint64, count) + if err := binary.Read(r, binary.LittleEndian, bits); err != nil { + return nil, err + } + v := make([]float64, count) + for i, b := range bits { + v[i] = math.Float64frombits(b) + } + return v, nil + default: + return nil, fmt.Errorf("pvdata: unknown array type code 0x%02X", tc) + } +} + +func readStruct(r io.Reader, desc FieldDesc) (StructValue, error) { + sv := StructValue{TypeID: desc.TypeID, Fields: make([]FieldValue, len(desc.Fields))} + for i, f := range desc.Fields { + v, err := ReadValue(r, f.Desc) + if err != nil { + return StructValue{}, fmt.Errorf("field %q: %w", f.Name, err) + } + sv.Fields[i] = FieldValue{Name: f.Name, Value: v} + } + return sv, nil +} + +func readStructArr(r io.Reader, desc FieldDesc) ([]StructValue, error) { + n, err := ReadSize(r) + if err != nil { + return nil, err + } + arr := make([]StructValue, n) + for i := range arr { + sv, err := readStruct(r, desc) + if err != nil { + return nil, err + } + arr[i] = sv + } + return arr, nil +} + +func readUnion(r io.Reader, desc FieldDesc) (UnionValue, error) { + idx, err := ReadSize(r) + if err != nil { + return UnionValue{}, err + } + if idx < 0 || int(idx) >= len(desc.Fields) { + return UnionValue{Selected: int(idx)}, nil // null selector + } + f := desc.Fields[idx] + v, err := ReadValue(r, f.Desc) + if err != nil { + return UnionValue{}, err + } + return UnionValue{Selected: int(idx), Name: f.Name, Value: v}, nil +} + +func readUnionArr(r io.Reader, desc FieldDesc) ([]UnionValue, error) { + n, err := ReadSize(r) + if err != nil { + return nil, err + } + arr := make([]UnionValue, n) + for i := range arr { + uv, err := readUnion(r, desc) + if err != nil { + return nil, err + } + arr[i] = uv + } + return arr, nil +} + +func readVariant(r io.Reader) (VariantValue, error) { + desc, err := ReadFieldDesc(r) + if err != nil { + return VariantValue{}, err + } + v, err := ReadValue(r, desc) + if err != nil { + return VariantValue{}, err + } + return VariantValue{Desc: desc, Value: v}, nil +} + +// WriteValue encodes v (which must match desc) into w. +func WriteValue(w io.Writer, desc FieldDesc, v Value) error { + switch desc.Kind { + case KindScalar: + return writeScalar(w, desc.TypeCode, v) + case KindScalarArr: + return writeScalarArray(w, desc.TypeCode, v) + case KindString: + return WriteString(w, v.(string)) + case KindStringArr: + ss := v.([]string) + if err := WriteSize(w, int64(len(ss))); err != nil { + return err + } + for _, s := range ss { + if err := WriteString(w, s); err != nil { + return err + } + } + return nil + case KindStruct: + return writeStruct(w, desc, v.(StructValue)) + case KindUnion: + return writeUnion(w, v.(UnionValue)) + case KindVariant: + vv := v.(VariantValue) + if err := WriteFieldDesc(w, vv.Desc); err != nil { + return err + } + return WriteValue(w, vv.Desc, vv.Value) + default: + return fmt.Errorf("pvdata: WriteValue unsupported kind %d", desc.Kind) + } +} + +func writeScalar(w io.Writer, tc byte, v Value) error { + switch tc { + case TypeCodeBoolean: + b := byte(0) + if v.(bool) { + b = 1 + } + _, err := w.Write([]byte{b}) + return err + case TypeCodeFloat: + return binary.Write(w, binary.LittleEndian, math.Float32bits(v.(float32))) + case TypeCodeDouble: + return binary.Write(w, binary.LittleEndian, math.Float64bits(v.(float64))) + default: + return binary.Write(w, binary.LittleEndian, v) + } +} + +func writeScalarArray(w io.Writer, tc byte, v Value) error { + switch tc { + case TypeCodeDoubleArr: + arr := v.([]float64) + if err := WriteSize(w, int64(len(arr))); err != nil { + return err + } + bits := make([]uint64, len(arr)) + for i, f := range arr { + bits[i] = math.Float64bits(f) + } + return binary.Write(w, binary.LittleEndian, bits) + case TypeCodeFloatArr: + arr := v.([]float32) + if err := WriteSize(w, int64(len(arr))); err != nil { + return err + } + bits := make([]uint32, len(arr)) + for i, f := range arr { + bits[i] = math.Float32bits(f) + } + return binary.Write(w, binary.LittleEndian, bits) + default: + // for integer arrays, v is already []intN or []uintN + if err := WriteSize(w, int64(arrayLen(v))); err != nil { + return err + } + return binary.Write(w, binary.LittleEndian, v) + } +} + +func writeStruct(w io.Writer, desc FieldDesc, sv StructValue) error { + for i, fv := range sv.Fields { + if err := WriteValue(w, desc.Fields[i].Desc, fv.Value); err != nil { + return fmt.Errorf("field %q: %w", fv.Name, err) + } + } + return nil +} + +func writeUnion(w io.Writer, uv UnionValue) error { + if err := WriteSize(w, int64(uv.Selected)); err != nil { + return err + } + // caller must supply desc to encode value; not needed if Selected is null + return nil +} + +// arrayLen returns reflect-free length for the supported array types. +func arrayLen(v Value) int { + switch x := v.(type) { + case []bool: + return len(x) + case []int8: + return len(x) + case []int16: + return len(x) + case []int32: + return len(x) + case []int64: + return len(x) + case []uint8: + return len(x) + case []uint16: + return len(x) + case []uint32: + return len(x) + case []uint64: + return len(x) + default: + return 0 + } +} diff --git a/pkg/pva/pvdata/types_test.go b/pkg/pva/pvdata/types_test.go new file mode 100644 index 0000000..d32901b --- /dev/null +++ b/pkg/pva/pvdata/types_test.go @@ -0,0 +1,220 @@ +package pvdata_test + +import ( + "bytes" + "testing" + + "github.com/uopi/gopva/pvdata" +) + +// ---- Compact size encoding ------------------------------------------- + +func TestCompactSize(t *testing.T) { + cases := []int64{0, 1, 10, 254, 255, 256, 1000, 65535, 70000, 1<<31 - 1} + for _, n := range cases { + var buf bytes.Buffer + if err := pvdata.WriteSize(&buf, n); err != nil { + t.Fatalf("WriteSize(%d): %v", n, err) + } + got, err := pvdata.ReadSize(&buf) + if err != nil { + t.Fatalf("ReadSize for %d: %v", n, err) + } + if got != n { + t.Errorf("round-trip size %d → %d", n, got) + } + } +} + +// ---- String encoding ------------------------------------------------ + +func TestString(t *testing.T) { + cases := []string{"", "hello", "UOPI:TICK", "a string with spaces and unicode: é"} + for _, s := range cases { + var buf bytes.Buffer + if err := pvdata.WriteString(&buf, s); err != nil { + t.Fatalf("WriteString(%q): %v", s, err) + } + got, err := pvdata.ReadString(&buf) + if err != nil { + t.Fatalf("ReadString for %q: %v", s, err) + } + if got != s { + t.Errorf("round-trip string %q → %q", s, got) + } + } +} + +// ---- FieldDesc (introspection) round-trip --------------------------- + +func TestFieldDescScalar(t *testing.T) { + desc := pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeDouble} + var buf bytes.Buffer + if err := pvdata.WriteFieldDesc(&buf, desc); err != nil { + t.Fatal(err) + } + got, err := pvdata.ReadFieldDesc(&buf) + if err != nil { + t.Fatal(err) + } + if got.TypeCode != desc.TypeCode || got.Kind != desc.Kind { + t.Errorf("scalar round-trip: got %+v, want %+v", got, desc) + } +} + +func TestFieldDescStruct(t *testing.T) { + // Minimal NTScalar-like struct: value(double) + timeStamp(struct{…}) + desc := pvdata.FieldDesc{ + Kind: pvdata.KindStruct, + TypeCode: pvdata.TypeCodeStruct, + TypeID: "epics:nt/NTScalar:1.0", + Fields: []pvdata.Field{ + {Name: "value", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeDouble}}, + {Name: "alarm", Desc: pvdata.FieldDesc{ + Kind: pvdata.KindStruct, TypeCode: pvdata.TypeCodeStruct, TypeID: "alarm_t", + Fields: []pvdata.Field{ + {Name: "severity", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeInt}}, + {Name: "status", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeInt}}, + {Name: "message", Desc: pvdata.FieldDesc{Kind: pvdata.KindString, TypeCode: pvdata.TypeCodeString}}, + }, + }}, + {Name: "timeStamp", Desc: pvdata.FieldDesc{ + Kind: pvdata.KindStruct, TypeCode: pvdata.TypeCodeStruct, TypeID: "time_t", + Fields: []pvdata.Field{ + {Name: "secondsPastEpoch", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeLong}}, + {Name: "nanoseconds", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeInt}}, + {Name: "userTag", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeInt}}, + }, + }}, + }, + } + + var buf bytes.Buffer + if err := pvdata.WriteFieldDesc(&buf, desc); err != nil { + t.Fatal(err) + } + got, err := pvdata.ReadFieldDesc(&buf) + if err != nil { + t.Fatal(err) + } + if got.TypeID != desc.TypeID { + t.Errorf("typeID: got %q want %q", got.TypeID, desc.TypeID) + } + if len(got.Fields) != len(desc.Fields) { + t.Fatalf("field count: got %d want %d", len(got.Fields), len(desc.Fields)) + } + for i, f := range desc.Fields { + if got.Fields[i].Name != f.Name { + t.Errorf("field[%d] name: got %q want %q", i, got.Fields[i].Name, f.Name) + } + } +} + +// ---- Value round-trip ----------------------------------------------- + +func TestValueDouble(t *testing.T) { + desc := pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeDouble} + var v pvdata.Value = float64(3.14159) + var buf bytes.Buffer + if err := pvdata.WriteValue(&buf, desc, v); err != nil { + t.Fatal(err) + } + got, err := pvdata.ReadValue(&buf, desc) + if err != nil { + t.Fatal(err) + } + if got.(float64) != v.(float64) { + t.Errorf("double round-trip: got %v want %v", got, v) + } +} + +func TestValueIntArray(t *testing.T) { + desc := pvdata.FieldDesc{Kind: pvdata.KindScalarArr, TypeCode: pvdata.TypeCodeIntArr} + var v pvdata.Value = []int32{1, 2, 3, -7, 1<<30} + var buf bytes.Buffer + if err := pvdata.WriteValue(&buf, desc, v); err != nil { + t.Fatal(err) + } + got, err := pvdata.ReadValue(&buf, desc) + if err != nil { + t.Fatal(err) + } + arr := got.([]int32) + src := v.([]int32) + if len(arr) != len(src) { + t.Fatalf("int32 array length: got %d want %d", len(arr), len(src)) + } + for i := range src { + if arr[i] != src[i] { + t.Errorf("int32[%d]: got %d want %d", i, arr[i], src[i]) + } + } +} + +func TestValueStruct(t *testing.T) { + desc := pvdata.FieldDesc{ + Kind: pvdata.KindStruct, + TypeCode: pvdata.TypeCodeStruct, + TypeID: "test_t", + Fields: []pvdata.Field{ + {Name: "x", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeDouble}}, + {Name: "n", Desc: pvdata.FieldDesc{Kind: pvdata.KindScalar, TypeCode: pvdata.TypeCodeInt}}, + {Name: "s", Desc: pvdata.FieldDesc{Kind: pvdata.KindString, TypeCode: pvdata.TypeCodeString}}, + }, + } + sv := pvdata.StructValue{ + TypeID: "test_t", + Fields: []pvdata.FieldValue{ + {Name: "x", Value: float64(2.718)}, + {Name: "n", Value: int32(42)}, + {Name: "s", Value: "hello"}, + }, + } + var buf bytes.Buffer + if err := pvdata.WriteValue(&buf, desc, sv); err != nil { + t.Fatal(err) + } + got, err := pvdata.ReadValue(&buf, desc) + if err != nil { + t.Fatal(err) + } + gsv := got.(pvdata.StructValue) + if gsv.Fields[0].Value.(float64) != sv.Fields[0].Value.(float64) { + t.Errorf("x field mismatch") + } + if gsv.Fields[1].Value.(int32) != sv.Fields[1].Value.(int32) { + t.Errorf("n field mismatch") + } + if gsv.Fields[2].Value.(string) != sv.Fields[2].Value.(string) { + t.Errorf("s field mismatch") + } +} + +// ---- BitSet --------------------------------------------------------- + +func TestBitSet(t *testing.T) { + var bs pvdata.BitSet + bs.Set(0) + bs.Set(3) + bs.Set(15) + bs.Set(16) + + var buf bytes.Buffer + if err := pvdata.WriteBitSet(&buf, bs); err != nil { + t.Fatal(err) + } + got, err := pvdata.ReadBitSet(&buf) + if err != nil { + t.Fatal(err) + } + for _, idx := range []int{0, 3, 15, 16} { + if !got.Has(idx) { + t.Errorf("BitSet missing index %d after round-trip", idx) + } + } + for _, idx := range []int{1, 2, 4, 14, 17} { + if got.Has(idx) { + t.Errorf("BitSet has unexpected index %d", idx) + } + } +} diff --git a/workspace/softioc.log b/workspace/softioc.log index e9cbd71..5e88df6 100644 --- a/workspace/softioc.log +++ b/workspace/softioc.log @@ -1,38 +1,2 @@ Starting iocInit iocRun: All initialization complete -CAS: request from 127.0.0.1:56104 => bad resource ID -CAS: Request from 127.0.0.1:56104 => cmmd=1 cid=0x2 type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56104 => available=0x1 N=0 paddr=(nil) -CAS: forcing disconnect from 127.0.0.1:56104 -CAS: request from 127.0.0.1:56106 => bad resource ID -CAS: Request from 127.0.0.1:56106 => cmmd=1 cid=0x5 type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56106 => available=0x3 N=0 paddr=(nil) -CAS: forcing disconnect from 127.0.0.1:56106 -CAS: request from 127.0.0.1:56120 => bad resource ID -CAS: Request from 127.0.0.1:56120 => cmmd=1 cid=0x8 type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56120 => available=0x6 N=0 paddr=(nil) -CAS: forcing disconnect from 127.0.0.1:56120 -CAS: request from 127.0.0.1:56126 => bad resource ID -CAS: Request from 127.0.0.1:56126 => cmmd=1 cid=0xb type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56126 => available=0xa N=0 paddr=(nil) -CAS: forcing disconnect from 127.0.0.1:56126 -CAS: request from 127.0.0.1:56130 => no memory to add subscription to db -CAS: Request from 127.0.0.1:56130 => cmmd=1 cid=0xe type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56130 => available=0xf N=0 paddr=0x7fae3800d838 -CAS: forcing disconnect from 127.0.0.1:56130 -CAS: request from 127.0.0.1:56132 => no memory to add subscription to db -CAS: Request from 127.0.0.1:56132 => cmmd=1 cid=0x10 type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56132 => available=0x11 N=0 paddr=0x7fae3800d768 -CAS: forcing disconnect from 127.0.0.1:56132 -CAS: request from 127.0.0.1:56144 => bad resource ID -CAS: Request from 127.0.0.1:56144 => cmmd=1 cid=0x10 type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56144 => available=0x17 N=0 paddr=(nil) -CAS: forcing disconnect from 127.0.0.1:56144 -CAS: request from 127.0.0.1:56148 => bad resource ID -CAS: Request from 127.0.0.1:56148 => cmmd=1 cid=0x10 type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:56148 => available=0x1d N=0 paddr=(nil) -CAS: forcing disconnect from 127.0.0.1:56148 -CAS: request from 127.0.0.1:60292 => bad resource ID -CAS: Request from 127.0.0.1:60292 => cmmd=1 cid=0x10 type=25 count=1 postsize=16 -CAS: Request from 127.0.0.1:60292 => available=0x24 N=0 paddr=(nil) -CAS: forcing disconnect from 127.0.0.1:60292