Remove proxy.NewBatchFactory; convert BatchFactory directly from node.Key

Author: pjdotson

core/pkg/distribution/channel/service.go

@@ -128,9 +128,9 @@ func OpenService(ctx context.Context, cfgs ...ServiceConfig) (s *Service, err er
 		db:           cfg.ClusterDB,
 		otg:          cfg.Ontology,
 		indexes:      newIndexes(),
-		createRouter: proxy.NewBatchFactory[Channel](cfg.HostResolver.HostKey()),
-		keyRouter:    proxy.NewBatchFactory[Key](cfg.HostResolver.HostKey()),
-		renameRouter: proxy.NewBatchFactory[renameBatchEntry](cfg.HostResolver.HostKey()),
+		createRouter: proxy.BatchFactory[Channel](cfg.HostResolver.HostKey()),
+		keyRouter:    proxy.BatchFactory[Key](cfg.HostResolver.HostKey()),
+		renameRouter: proxy.BatchFactory[renameBatchEntry](cfg.HostResolver.HostKey()),
 	}
 	cleanup, ok := service.NewOpener(ctx, &s.closer)
 	defer func() { err = cleanup(err) }()

core/pkg/distribution/framer/deleter/lease_proxy.go

@@ -32,7 +32,7 @@ func newLeaseProxy(
 ) (*leaseProxy, error) {
 	p := &leaseProxy{
 		ServiceConfig: cfg,
-		keyRouter:     proxy.NewBatchFactory[channel.Key](cfg.HostResolver.HostKey()),
+		keyRouter:     proxy.BatchFactory[channel.Key](cfg.HostResolver.HostKey()),
 	}
 	return p, nil
 }

core/pkg/distribution/framer/iterator/service.go

@@ -157,7 +157,7 @@ func (s *Service) NewStream(ctx context.Context, cfg Config) (StreamIterator, er
 	cfg.Keys = cfg.Keys.Unique()
 	var (
 		hostID             = s.cfg.HostResolver.HostKey()
-		batch              = proxy.NewBatchFactory[channel.Key](hostID).Batch(cfg.Keys)
+		batch              = proxy.BatchFactory[channel.Key](hostID).Batch(cfg.Keys)
 		pipe               = plumber.New()
 		needPeerRouting    = len(batch.Peers) > 0
 		needGatewayRouting = len(batch.Gateway) > 0

core/pkg/distribution/framer/writer/service.go

@@ -351,7 +351,7 @@ func (s *Service) NewStream(ctx context.Context, cfgs ...Config) (StreamWriter,
 
 	var (
 		hostKey           = s.cfg.HostResolver.HostKey()
-		batch             = proxy.NewBatchFactory[keyAuthority](hostKey).Batch(cfg.keyAuthorities())
+		batch             = proxy.BatchFactory[keyAuthority](hostKey).Batch(cfg.keyAuthorities())
 		pipe              = plumber.New()
 		hasPeer           = len(batch.Peers) > 0
 		hasGateway        = len(batch.Gateway) > 0

core/pkg/distribution/proxy/proxy.go

@@ -23,15 +23,10 @@ type Entry interface {
 }
 
 // BatchFactory partitions Entry values by their leaseholder relative to a fixed host
-// node. Construct one with NewBatchFactory, then call Batch.
-type BatchFactory[E Entry] struct{ host node.Key }
-
-// NewBatchFactory returns a BatchFactory that treats host as the local node: Batch
-// routes entries leased to host into the gateway bucket and all other leased entries to
-// peers.
-func NewBatchFactory[E Entry](host node.Key) BatchFactory[E] {
-	return BatchFactory[E]{host: host}
-}
+// node, treating that host as the local node: Batch routes entries leased to it into the
+// gateway bucket and all other leased entries to peers. Construct one by converting a
+// host node.Key, e.g. BatchFactory[E](host), then call Batch.
+type BatchFactory[E Entry] node.Key
 
 // Batch is the result of partitioning a set of Entry values by leaseholder. Every input
 // entry appears in exactly one of the three buckets.
@@ -50,12 +45,13 @@ type Batch[E Entry] struct {
 // grouped into Peers by their leaseholder key. The relative order of entries is
 // preserved within each bucket.
 func (f BatchFactory[E]) Batch(entries []E) Batch[E] {
+	host := node.Key(f)
 	b := Batch[E]{Peers: make(map[node.Key][]E)}
 	for _, entry := range entries {
 		lease := entry.Lease()
 		if lease.IsFree() {
 			b.Free = append(b.Free, entry)
-		} else if lease == f.host {
+		} else if lease == host {
 			b.Gateway = append(b.Gateway, entry)
 		} else {
 			b.Peers[lease] = append(b.Peers[lease], entry)

core/pkg/distribution/proxy/proxy_test.go

@@ -34,7 +34,7 @@ func (e entry) Lease() node.Key { return e.lease }
 var _ = Describe("Batch", func() {
 	var factory proxy.BatchFactory[entry]
 	BeforeEach(func() {
-		factory = proxy.NewBatchFactory[entry](host)
+		factory = proxy.BatchFactory[entry](host)
 	})
 
 	It("Should route entries leased to the host into the gateway bucket", func() {