Go client: port bulk read/write SDK methods + CLI subcommands

Mirrors the .NET addition: HEAD's session.go had only the subscribe-style bulks (AddItemBulk / AdviseItemBulk / RemoveItemBulk / UnAdviseItemBulk / SubscribeBulk / UnsubscribeBulk). This commit ports the value-bulk SDK surface and CLI subcommands from divergent branch commit f220908. SDK (clients/go/mxgateway/session.go): - WriteBulk(ctx, serverHandle int32, entries []*WriteBulkEntry) - Write2Bulk(ctx, ..., entries []*Write2BulkEntry) - WriteSecuredBulk(ctx, ..., entries []*WriteSecuredBulkEntry) - WriteSecured2Bulk(ctx, ..., entries []*WriteSecured2BulkEntry) - ReadBulk(ctx, serverHandle int32, tagAddresses []string, timeout time.Duration) → []*BulkReadResult types.go gains public re-exports of the generated proto types (WriteBulkCommand, WriteBulkEntry, Write2BulkCommand, Write2BulkEntry, WriteSecuredBulkCommand, WriteSecuredBulkEntry, WriteSecured2BulkCommand, WriteSecured2BulkEntry, ReadBulkCommand, BulkWriteReply, BulkWriteResult, BulkReadReply, BulkReadResult) so external callers can construct entries through the public `mxgateway` package without dipping into the internal generated path. CLI (clients/go/cmd/mxgw-go/main.go): - read-bulk, write-bulk, write2-bulk, write-secured-bulk, write-secured2-bulk routed through runWithIO. write families share a runWriteBulkVariant helper that gates per-variant flags (--current-user-id, --verifier-user-id, --timestamp) so the Client.Go-015 flag-gating contract is preserved. - bench-read-bulk: percentile + timing helpers; JSON output schema identical to the .NET / Rust / Python / Java benches. parseInt32List was changed from panic-on-error to ([]int32, error) so the new write-bulk commands surface parse errors gracefully; the existing runUnsubscribeBulk caller is updated accordingly. Verification: go build ./... + go vet ./... + go test ./... all clean. Manual smoke against live gateway on localhost:5120: open-session → register → subscribe-bulk on 3 TestMachine_NNN.TestChangingInt tags (all wasSuccessful=true) → read-bulk (all wasSuccessful=true / wasCached=true) → write-bulk int32 100/200/300 (all wasSuccessful=true) → close-session SESSION_STATE_CLOSED. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-24 04:49:33 -04:00
parent b3ae200b11
commit 8aaab82287
3 changed files with 563 additions and 5 deletions
@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"time"

 	pb "gitea.dohertylan.com/dohertj2/mxaccessgw/clients/go/internal/generated"
 	"google.golang.org/grpc/codes"
@@ -387,6 +388,142 @@ func (s *Session) UnsubscribeBulk(ctx context.Context, serverHandle int32, itemH
 	return reply.GetUnsubscribeBulk().GetResults(), nil
 }

+// WriteBulk invokes MXAccess Write sequentially for each entry inside one gateway command.
+// Per-entry failures appear as BulkWriteResult entries with WasSuccessful=false; the call
+// never returns an error for per-entry MXAccess failures (it returns an error only for
+// protocol-level failures or transport errors).
+func (s *Session) WriteBulk(ctx context.Context, serverHandle int32, entries []*WriteBulkEntry) ([]*BulkWriteResult, error) {
+	if entries == nil {
+		return nil, errors.New("mxgateway: write bulk entries are required")
+	}
+	if err := ensureBulkSize("write bulk entries", len(entries)); err != nil {
+		return nil, err
+	}
+	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
+		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_BULK,
+		Payload: &pb.MxCommand_WriteBulk{
+			WriteBulk: &pb.WriteBulkCommand{
+				ServerHandle: serverHandle,
+				Entries:      entries,
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return reply.GetWriteBulk().GetResults(), nil
+}
+
+// Write2Bulk invokes MXAccess Write2 (timestamped) for each entry inside one gateway command.
+func (s *Session) Write2Bulk(ctx context.Context, serverHandle int32, entries []*Write2BulkEntry) ([]*BulkWriteResult, error) {
+	if entries == nil {
+		return nil, errors.New("mxgateway: write2 bulk entries are required")
+	}
+	if err := ensureBulkSize("write2 bulk entries", len(entries)); err != nil {
+		return nil, err
+	}
+	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
+		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE2_BULK,
+		Payload: &pb.MxCommand_Write2Bulk{
+			Write2Bulk: &pb.Write2BulkCommand{
+				ServerHandle: serverHandle,
+				Entries:      entries,
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return reply.GetWrite2Bulk().GetResults(), nil
+}
+
+// WriteSecuredBulk invokes MXAccess WriteSecured for each entry. Credential-sensitive
+// values must not be logged by callers; mirrors the single-item WriteSecured contract.
+func (s *Session) WriteSecuredBulk(ctx context.Context, serverHandle int32, entries []*WriteSecuredBulkEntry) ([]*BulkWriteResult, error) {
+	if entries == nil {
+		return nil, errors.New("mxgateway: write-secured bulk entries are required")
+	}
+	if err := ensureBulkSize("write-secured bulk entries", len(entries)); err != nil {
+		return nil, err
+	}
+	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
+		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_SECURED_BULK,
+		Payload: &pb.MxCommand_WriteSecuredBulk{
+			WriteSecuredBulk: &pb.WriteSecuredBulkCommand{
+				ServerHandle: serverHandle,
+				Entries:      entries,
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return reply.GetWriteSecuredBulk().GetResults(), nil
+}
+
+// WriteSecured2Bulk invokes MXAccess WriteSecured2 (timestamped) for each entry.
+func (s *Session) WriteSecured2Bulk(ctx context.Context, serverHandle int32, entries []*WriteSecured2BulkEntry) ([]*BulkWriteResult, error) {
+	if entries == nil {
+		return nil, errors.New("mxgateway: write-secured2 bulk entries are required")
+	}
+	if err := ensureBulkSize("write-secured2 bulk entries", len(entries)); err != nil {
+		return nil, err
+	}
+	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
+		Kind: pb.MxCommandKind_MX_COMMAND_KIND_WRITE_SECURED2_BULK,
+		Payload: &pb.MxCommand_WriteSecured2Bulk{
+			WriteSecured2Bulk: &pb.WriteSecured2BulkCommand{
+				ServerHandle: serverHandle,
+				Entries:      entries,
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return reply.GetWriteSecured2Bulk().GetResults(), nil
+}
+
+// ReadBulk snapshots the current value of each requested tag.
+//
+// MXAccess COM has no synchronous Read; the worker satisfies this by returning the
+// most recent cached OnDataChange value when the tag is already advised (WasCached=true),
+// or by taking a full AddItem + Advise + wait + UnAdvise + RemoveItem snapshot lifecycle
+// otherwise. timeout bounds the wait per tag in the snapshot case; pass zero to use the
+// worker default. Per-tag failures (timeout, invalid tag) appear as BulkReadResult entries
+// with WasSuccessful=false; the call never returns an error for per-tag MXAccess failures.
+func (s *Session) ReadBulk(ctx context.Context, serverHandle int32, tagAddresses []string, timeout time.Duration) ([]*BulkReadResult, error) {
+	if tagAddresses == nil {
+		return nil, errors.New("mxgateway: tag addresses are required")
+	}
+	if err := ensureBulkSize("tag addresses", len(tagAddresses)); err != nil {
+		return nil, err
+	}
+	var timeoutMs uint32
+	if timeout > 0 {
+		ms := timeout.Milliseconds()
+		if ms > int64(^uint32(0)) {
+			timeoutMs = ^uint32(0)
+		} else {
+			timeoutMs = uint32(ms)
+		}
+	}
+	reply, err := s.invokeCommand(ctx, &pb.MxCommand{
+		Kind: pb.MxCommandKind_MX_COMMAND_KIND_READ_BULK,
+		Payload: &pb.MxCommand_ReadBulk{
+			ReadBulk: &pb.ReadBulkCommand{
+				ServerHandle: serverHandle,
+				TagAddresses: tagAddresses,
+				TimeoutMs:    timeoutMs,
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return reply.GetReadBulk().GetResults(), nil
+}
+
 // Write invokes MXAccess Write.
 func (s *Session) Write(ctx context.Context, serverHandle, itemHandle int32, value *MxValue, userID int32) error {
 	_, err := s.WriteRaw(ctx, serverHandle, itemHandle, value, userID)