rangefeed: add newBaseRegistration

Epic: none
Release note: None

Author: stevendanna

pkg/kv/kvserver/rangefeed/buffered_registration.go

@@ -78,25 +78,23 @@ func newBufferedRegistration(
 	withDiff bool,
 	withFiltering bool,
 	withOmitRemote bool,
-	withBulkDelivery int,
+	bulkDeliverySize int,
 	bufferSz int,
 	blockWhenFull bool,
 	metrics *Metrics,
 	stream Stream,
 	removeRegFromProcessor func(registration),
 ) *bufferedRegistration {
 	br := &bufferedRegistration{
-		baseRegistration: baseRegistration{
-			streamCtx:              streamCtx,
-			span:                   span,
-			keys:                   span.AsRange(),
-			catchUpTimestamp:       startTS,
-			withDiff:               withDiff,
-			withFiltering:          withFiltering,
-			withOmitRemote:         withOmitRemote,
-			removeRegFromProcessor: removeRegFromProcessor,
-			bulkDelivery:           withBulkDelivery,
-		},
+		baseRegistration: newBaseRegistration(
+			streamCtx,
+			span,
+			startTS,
+			withDiff,
+			withFiltering,
+			withOmitRemote,
+			bulkDeliverySize,
+			removeRegFromProcessor),
 		metrics:       metrics,
 		stream:        stream,
 		buf:           make(chan *sharedEvent, bufferSz),

pkg/kv/kvserver/rangefeed/registry.go

@@ -105,6 +105,32 @@ type baseRegistration struct {
 	shouldUnreg atomic.Bool
 }
 
+// newBaseRegistration returns a baseRegistration. Note that a baseRegistration
+// does not implement registration. Rather this is used by the buffered and
+// unbuffered registrations to share some basic methods.
+func newBaseRegistration(
+	streamCtx context.Context,
+	span roachpb.Span,
+	startTS hlc.Timestamp,
+	withDiff bool,
+	withFiltering bool,
+	withOmitRemote bool,
+	bulkDeliverySize int,
+	removeRegFromProcessor func(registration),
+) baseRegistration {
+	return baseRegistration{
+		streamCtx:              streamCtx,
+		span:                   span,
+		keys:                   span.AsRange(),
+		catchUpTimestamp:       startTS,
+		withDiff:               withDiff,
+		withFiltering:          withFiltering,
+		withOmitRemote:         withOmitRemote,
+		bulkDelivery:           bulkDeliverySize,
+		removeRegFromProcessor: removeRegFromProcessor,
+	}
+}
+
 // ID implements interval.Interface.
 func (r *baseRegistration) ID() uintptr {
 	return uintptr(r.id)

pkg/kv/kvserver/rangefeed/unbuffered_registration.go

@@ -102,17 +102,15 @@ func newUnbufferedRegistration(
 	removeRegFromProcessor func(registration),
 ) *unbufferedRegistration {
 	br := &unbufferedRegistration{
-		baseRegistration: baseRegistration{
-			streamCtx:              streamCtx,
-			span:                   span,
-			keys:                   span.AsRange(),
-			catchUpTimestamp:       startTS,
-			withDiff:               withDiff,
-			withFiltering:          withFiltering,
-			withOmitRemote:         withOmitRemote,
-			removeRegFromProcessor: removeRegFromProcessor,
-			bulkDelivery:           bulkDeliverySize,
-		},
+		baseRegistration: newBaseRegistration(
+			streamCtx,
+			span,
+			startTS,
+			withDiff,
+			withFiltering,
+			withOmitRemote,
+			bulkDeliverySize,
+			removeRegFromProcessor),
 		metrics: metrics,
 		stream:  stream,
 	}