api: update default read and write buffer size to 32

api/dataplane/v1/edge/edge_end.go

@@ -14,13 +14,16 @@ type edgeEnd struct {
 
 func newEdgeEnd(dialer client.Dialer, opts ...EdgeOption) (*edgeEnd, error) {
 	// options
-	eopt := &edgeOption{}
+	eopt := &edgeOption{
+		readBufferSize:  -1,
+		writeBufferSize: -1,
+	}
 	for _, opt := range opts {
 		opt(eopt)
 	}
-	eopts := &client.RetryEndOptions{
-		EndOptions: &client.EndOptions{},
-	}
+
+	// turn to end options
+	eopts := client.NewEndOptions()
 	if eopt.tmr != nil {
 		eopts.SetTimer(eopt.tmr)
 	}
@@ -33,6 +36,9 @@ func newEdgeEnd(dialer client.Dialer, opts ...EdgeOption) (*edgeEnd, error) {
 	if eopt.meta != nil {
 		eopts.SetMeta(eopt.meta)
 	}
+	if eopt.readBufferSize != -1 || eopt.writeBufferSize != -1 {
+		eopts.SetBufferSize(eopt.readBufferSize, eopt.writeBufferSize)
+	}
 
 	// new geminio end
 	end, err := client.NewRetryEndWithDialer(dialer, eopts)

api/dataplane/v1/edge/edge_option.go

@@ -10,10 +10,11 @@ type Logger log.Logger
 type Timer timer.Timer
 
 type edgeOption struct {
-	logger Logger
-	tmr    Timer
-	edgeID *uint64
-	meta   []byte
+	logger                          Logger
+	tmr                             Timer
+	edgeID                          *uint64
+	meta                            []byte
+	readBufferSize, writeBufferSize int
 }
 
 type EdgeOption func(*edgeOption)
@@ -43,3 +44,10 @@ func OptionEdgeMeta(meta []byte) EdgeOption {
 		opt.meta = meta
 	}
 }
+
+func OptionServiceBufferSize(read, write int) EdgeOption {
+	return func(opt *edgeOption) {
+		opt.readBufferSize = read
+		opt.writeBufferSize = write
+	}
+}

api/dataplane/v1/service/service_end.go

@@ -18,7 +18,10 @@ type serviceEnd struct {
 
 func newServiceEnd(dialer client.Dialer, opts ...ServiceOption) (*serviceEnd, error) {
 	// options
-	sopt := &serviceOption{}
+	sopt := &serviceOption{
+		readBufferSize:  -1, // default length 128
+		writeBufferSize: -1, // default legnth 128
+	}
 	for _, opt := range opts {
 		opt(sopt)
 	}
@@ -32,6 +35,9 @@ func newServiceEnd(dialer client.Dialer, opts ...ServiceOption) (*serviceEnd, er
 	if sopt.serviceID != 0 {
 		sopts.SetClientID(sopt.serviceID)
 	}
+	if sopt.readBufferSize != -1 || sopt.writeBufferSize != -1 {
+		sopts.SetBufferSize(sopt.readBufferSize, sopt.writeBufferSize)
+	}
 	// meta
 	meta := &apis.Meta{}
 	if sopt.topics != nil {
@@ -60,13 +66,16 @@ func newServiceEnd(dialer client.Dialer, opts ...ServiceOption) (*serviceEnd, er
 
 func newRetryServiceEnd(dialer client.Dialer, opts ...ServiceOption) (*serviceEnd, error) {
 	// options
-	sopt := &serviceOption{}
+	sopt := &serviceOption{
+		readBufferSize:  -1,
+		writeBufferSize: -1,
+	}
 	for _, opt := range opts {
 		opt(sopt)
 	}
-	sopts := &client.RetryEndOptions{
-		EndOptions: &client.EndOptions{},
-	}
+
+	// turn to end options
+	sopts := client.NewEndOptions()
 	if sopt.tmr != nil {
 		sopts.SetTimer(sopt.tmr)
 	}
@@ -91,6 +100,9 @@ func newRetryServiceEnd(dialer client.Dialer, opts ...ServiceOption) (*serviceEn
 	if sopt.delegate != nil {
 		sopts.SetDelegate(sopt.delegate)
 	}
+	if sopt.readBufferSize != -1 || sopt.writeBufferSize != -1 {
+		sopts.SetBufferSize(sopt.readBufferSize, sopt.writeBufferSize)
+	}
 	// new geminio end
 	end, err := client.NewRetryEndWithDialer(dialer, sopts)
 	if err != nil {

api/dataplane/v1/service/service_option.go

@@ -20,8 +20,9 @@ type serviceOption struct {
 	// to tell frontier what service we are
 	service string
 	// delegate to know online offline stuff
-	delegate  Delegate
-	serviceID uint64
+	delegate                        Delegate
+	serviceID                       uint64
+	readBufferSize, writeBufferSize int
 }
 
 type ServiceOption func(*serviceOption)
@@ -63,3 +64,10 @@ func OptionServiceID(serviceID uint64) ServiceOption {
 		opt.serviceID = serviceID
 	}
 }
+
+func OptionServiceBufferSize(read, write int) ServiceOption {
+	return func(opt *serviceOption) {
+		opt.readBufferSize = read
+		opt.writeBufferSize = write
+	}
+}