gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(rpc2): add testing of example

Browse Source 
. #22505814

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2024-08-15 17:47:44 +08:00
parent b7ea90c5b6
commit 07dcc69dff
20 changed files with 1498 additions and 191 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
blobstore/common/rpc2/body.go
View File

@ -28,17 +28,18 @@ type bodyAndTrailer struct {
req *Request
trailerOnce sync.Once
trailer *FixedHeader
closeOnce sync.Once
}
func (r *bodyAndTrailer) tryReadTrailer() error {
if r.remain < 0 {
panic("rpc2: response body read too much")
panic("rpc2: body read too much, remain < 0")
}
var err error
if r.remain == 0 { // try to read trailer
r.trailerOnce.Do(func() {
_, err = r.req.Trailer.ReadFrom(r.sr)
_, err = r.trailer.ReadFrom(r.sr)
})
}
return err
@ -65,15 +66,19 @@ func (r *bodyAndTrailer) WriteTo(w io.Writer) (int64, error) {
return 0, io.EOF
}
if lw, ok := w.(*LimitedWriter); !ok {
lw, ok := w.(*LimitedWriter)
if !ok {
return 0, ErrLimitedWriter
} else if lw.a > int64(r.remain) {
}
if lw.a > int64(r.remain) {
return 0, io.ErrShortWrite
}
n, err := r.br.WriteTo(w)
_, err := r.br.WriteTo(lw)
if err == errLimitedWrite {
err = nil
}
n := lw.a - lw.n // actual read bytes
r.remain -= int(n)
if err == nil {
err = r.tryReadTrailer()
@ -100,7 +105,9 @@ func (r *bodyAndTrailer) Close() (err error) {
}
// makeBodyWithTrailer body and trailer remain.
func makeBodyWithTrailer(sr *transport.SizedReader, req *Request, l int64, decode bool) Body {
func makeBodyWithTrailer(sr *transport.SizedReader, req *Request,
trailer *FixedHeader, l int64, decode bool,
) Body {
var br Body
if decode {
br = newEdBody(*req.checksum, sr, int(l), false)
@ -108,10 +115,11 @@ func makeBodyWithTrailer(sr *transport.SizedReader, req *Request, l int64, decod
br = sr
}
r := &bodyAndTrailer{
sr: sr,
br: br,
remain: int(l),
req: req,
sr: sr,
br: br,
remain: int(l),
req: req,
trailer: trailer,
}
r.tryReadTrailer()
return r

72
blobstore/common/rpc2/body_test.go Normal file
View File

@ -0,0 +1,72 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"io"
"testing"
"github.com/stretchr/testify/require"
)
func TestRpc2Body(t *testing.T) {
var b bodyAndTrailer
b.remain = -1
require.Panics(t, func() { b.tryReadTrailer() })
b.remain = 0
_, err := b.WriteTo(io.Discard)
require.ErrorIs(t, io.EOF, err)
b.remain = 1
_, err = b.WriteTo(io.Discard)
require.ErrorIs(t, ErrLimitedWriter, err)
_, err = b.WriteTo(LimitWriter(io.Discard, 2))
require.ErrorIs(t, io.ErrShortWrite, err)
}
func TestRpc2ReadFrame(t *testing.T) {
addr, cli, shutdown := newTcpServer()
defer shutdown()
cli.connector = defaultConnector(cli.ConnectorConfig)
{
conn, err := cli.connector.Get(testCtx, addr)
require.NoError(t, err)
frame, _ := conn.AllocFrame(1)
frame.Write([]byte{0xee})
conn.WriteFrame(frame)
_, err = conn.ReadFrame()
require.ErrorIs(t, io.EOF, err)
}
{
conn, err := cli.connector.Get(testCtx, addr)
require.NoError(t, err)
frame, _ := conn.AllocFrame(5)
frame.Write([]byte{0x1, 0x00, 0x00, 0x00})
conn.WriteFrame(frame)
_, err = conn.ReadFrame()
require.ErrorIs(t, io.EOF, err)
}
{
conn, err := cli.connector.Get(testCtx, addr)
require.NoError(t, err)
frame, _ := conn.AllocFrame(5)
frame.Write([]byte{0x1, 0x00, 0x00, 0x00, 0xee})
conn.WriteFrame(frame)
_, err = conn.ReadFrame()
require.ErrorIs(t, io.EOF, err)
}
}

63
blobstore/common/rpc2/client_test.go Normal file
View File

@ -0,0 +1,63 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"bytes"
"fmt"
"io"
"strings"
"testing"
"github.com/stretchr/testify/require"
)
type errNoParameter struct{ Codec }
func (errNoParameter) Marshal() ([]byte, error) { return nil, fmt.Errorf("codec") }
func (errNoParameter) Unmarshal([]byte) error { return fmt.Errorf("codec") }
func TestClientRetry(t *testing.T) {
var emptyCli Client
emptyCli.Close()
addr, cli, shutdown := newTcpServer()
defer shutdown()
for _, r := range []io.Reader{
bytes.NewBuffer([]byte("error")),
bytes.NewReader([]byte("error")),
strings.NewReader("error"),
} {
req, err := NewRequest(testCtx, addr, "/error", nil, r)
require.NoError(t, err)
require.Error(t, cli.DoWith(req, nil))
req, err = NewRequest(testCtx, addr, "/error", nil, r)
require.NoError(t, err)
req.GetBody = func() (io.ReadCloser, error) { return nil, fmt.Errorf("error") }
require.Error(t, cli.DoWith(req, nil))
}
}
func TestClientCodec(t *testing.T) {
addr, cli, shutdown := newTcpServer()
defer shutdown()
_, err := NewRequest(testCtx, addr, "/", errNoParameter{NoParameter}, nil)
require.Error(t, err)
req, err := NewRequest(testCtx, addr, "/", nil, bytes.NewReader(make([]byte, 4)))
require.NoError(t, err)
require.Error(t, cli.DoWith(req, errNoParameter{NoParameter}))
}

3
blobstore/common/rpc2/connector.go
View File

@ -16,6 +16,7 @@ package rpc2
import (
"context"
"errors"
"io"
"net"
"sync"
@ -56,7 +57,7 @@ func (t tcpDialer) Dial(ctx context.Context, addr string) (Conn, error) {
type rdmaDialer struct{}
func (rdmaDialer) Dial(ctx context.Context, addr string) (Conn, error) {
panic("rpc2: rdma not implements")
return nil, errors.New("rpc2: rdma not implements")
}
type Connector interface {

41
blobstore/common/rpc2/connector_test.go Normal file
View File

@ -0,0 +1,41 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestConn(t *testing.T) {
{
var conn tcpConn
require.Nil(t, conn.Allocator())
require.Panics(t, func() { conn.ReadOnce() })
}
{
var conf ConnectorConfig
conf.Network = "xxx"
require.Panics(t, func() { defaultConnector(conf) })
}
{
var conf ConnectorConfig
conf.Network = "rdma"
c := defaultConnector(conf)
_, err := c.Get(testCtx, "")
require.Error(t, err)
}
}

243
blobstore/common/rpc2/example_test.go Normal file
View File

@ -0,0 +1,243 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"bytes"
"crypto/md5"
crand "crypto/rand"
"encoding/hex"
"fmt"
"io"
mrand "math/rand"
)
func ExampleServer_base() {
addr, cli, shutdown := newTcpServer()
defer shutdown()
// Request & Response
req, _ := NewRequest(testCtx, addr, "/", nil, nil)
if err := cli.DoWith(req, nil); err != nil {
fmt.Println(err)
}
// Stream
sreq, _ := NewStreamRequest(testCtx, addr, "/stream", nil)
sc := StreamClient[noneCodec, noneCodec]{cli}
scli, _ := sc.Streaming(sreq, nil)
waitc := make(chan struct{})
go func() {
for {
_, err := scli.Recv()
if err == io.EOF {
close(waitc)
return
}
if err != nil {
fmt.Println(err)
}
}
}()
for range [10]struct{}{} {
if err := scli.Send(&noneCodec{}); err != nil {
fmt.Println(err)
}
}
scli.CloseSend()
<-waitc
// Output:
}
type strMessage struct{ str string }
func (s *strMessage) Size() int { return len(s.str) }
func (s *strMessage) Marshal() ([]byte, error) { return []byte(s.str), nil }
func (s *strMessage) MarshalTo(b []byte) (int, error) { return copy(b, []byte(s.str)), nil }
func (s *strMessage) Unmarshal(b []byte) error { s.str = string(b); return nil }
func handleMessage(w ResponseWriter, req *Request) error {
var args strMessage
req.ParseParameter(&args)
req.Body.Close()
args.str = "-> " + args.str
return w.WriteOK(&args)
}
func ExampleServer_request_message() {
handler := &Router{}
handler.Register("/", handleMessage)
server, cli, shutdown := newServer("tcp", handler)
defer shutdown()
args := &strMessage{str: "request message"}
// message in request & response body
req, _ := NewRequest(testCtx, server.Name, "/", nil, Codec2Reader(args))
if err := cli.DoWith(req, args); err != nil {
fmt.Println(err)
}
fmt.Println(args.str)
// Output:
// -> request message
}
func handleUpload(w ResponseWriter, req *Request) error {
var args strMessage
req.ParseParameter(&args)
hasher := req.checksum.Hasher()
if _, err := req.Body.WriteTo(LimitWriter(hasher, req.ContentLength)); err != nil {
return err
}
args.str = fmt.Sprint(req.checksum.Readable(hasher.Sum(nil)))
req.Body.Close()
return w.WriteOK(&args)
}
func ExampleServer_request_upload() {
handler := &Router{}
handler.Register("/", handleUpload)
server, cli, shutdown := newServer("tcp", handler)
defer shutdown()
args := &strMessage{str: "request data"}
buff := make([]byte, mrand.Intn(4<<20)+1<<20)
crand.Read(buff)
// request message in parameter & response message in body
req, _ := NewRequest(testCtx, server.Name, "/", args, bytes.NewReader(buff))
req.OptionCrcUpload()
req.ContentLength = int64(len(buff))
if err := cli.DoWith(req, args); err != nil {
fmt.Println(err)
}
hasher := req.checksum.Hasher()
hasher.Write(buff)
fmt.Println(args.str == fmt.Sprint(req.checksum.Readable(hasher.Sum(nil))))
// Output:
// true
}
func handleUpDown(w ResponseWriter, req *Request) error {
var args strMessage
req.ParseParameter(&args)
uhasher := req.checksum.Hasher()
if _, err := req.Body.WriteTo(LimitWriter(uhasher, req.ContentLength)); err != nil {
return err
}
req.Body.Close()
buff := make([]byte, mrand.Intn(4<<20)+1<<20)
crand.Read(buff)
dhasher := req.checksum.Hasher()
dhasher.Write(buff)
rr := req.checksum.Readable
args.str = fmt.Sprintf("%v %v", rr(uhasher.Sum(nil)), rr(dhasher.Sum(nil)))
w.SetContentLength(int64(len(buff)))
w.WriteHeader(200, &args)
_, err := w.ReadFrom(bytes.NewReader(buff))
return err
}
func ExampleServer_request_updown() {
handler := &Router{}
handler.Register("/", handleUpDown)
server, cli, shutdown := newServer("tcp", handler)
defer shutdown()
args := &strMessage{str: "upload & download"}
buff := make([]byte, mrand.Intn(4<<20)+1<<20)
crand.Read(buff)
// request & response message in parameter
req, _ := NewRequest(testCtx, server.Name, "/", args, bytes.NewReader(buff))
req.Option(func(r *Request) { _ = req.RequestHeader.ToString() })
req.OptionCrc()
req.ContentLength = int64(len(buff))
uhasher := req.checksum.Hasher()
uhasher.Write(buff)
dhasher := req.checksum.Hasher()
resp, _ := cli.Do(req, args)
defer resp.Body.Close()
b := make([]byte, 32<<20)
for {
n, err := resp.Body.Read(b)
dhasher.Write(b[:n])
if err == io.EOF {
break
}
}
rr := req.checksum.Readable
fmt.Println(args.str == fmt.Sprintf("%v %v", rr(uhasher.Sum(nil)), rr(dhasher.Sum(nil))))
// Output:
// true
}
func handleTrailer(w ResponseWriter, req *Request) error {
hasher := md5.New()
req.Body.WriteTo(LimitWriter(hasher, req.ContentLength))
req.Body.Close()
w.Header().Merge(req.Header)
w.Header().Add("add", "header-stable")
w.Trailer().Add("add", "trailer-stable")
w.Trailer().SetLen("md5", 32)
w.AfterBody(func() error { return nil })
w.AfterBody(func() error {
w.Trailer().Set("md5", hex.EncodeToString(hasher.Sum(nil)))
return nil
})
return w.WriteOK(nil)
}
func ExampleServer_trailer() {
handler := &Router{}
handler.Register("/", handleTrailer)
server, cli, shutdown := newServer("tcp", handler)
defer shutdown()
hasher := md5.New()
buff := make([]byte, mrand.Intn(4<<20)+1<<20)
crand.Read(buff)
req, _ := NewRequest(testCtx, server.Name, "/", nil,
io.TeeReader(bytes.NewReader(buff), hasher))
req.ContentLength = int64(len(buff))
req.Header.Add("add", "x")
req.Trailer.SetLen("md5", 32)
req.AfterBody = func() error {
req.Trailer.Set("md5", hex.EncodeToString(hasher.Sum(nil)))
return nil
}
resp, _ := cli.Do(req, nil)
resp.Body.Close()
fmt.Println(resp.Header.Get("add"))
fmt.Println(resp.Trailer.Get("add"))
fmt.Println(hex.EncodeToString(hasher.Sum(nil)) == resp.Trailer.Get("md5"))
// Output:
// header-stable
// trailer-stable
// true
}

11
blobstore/common/rpc2/header.go
View File

@ -86,15 +86,6 @@ func (h *Header) Merge(other Header) {
}
}
// ToFixedHeader copy to unstabled fixed header
func (h *Header) ToFixedHeader() FixedHeader {
fh := FixedHeader{}
for key, val := range h.M {
fh.Set(key, val)
}
return fh
}
func (fh *FixedHeader) newIfNil() {
if fh.M == nil {
fh.M = make(map[string]FixedValue)
@ -163,7 +154,7 @@ func (fh *FixedHeader) MergeHeader(h Header) {
func (fh *FixedHeader) AllSize() (n int) {
for _, v := range fh.M {
n += int(v.GetLen())
n += int(v.Len)
}
return
}

125
blobstore/common/rpc2/header_test.go Normal file
View File

@ -0,0 +1,125 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"io"
"strconv"
"testing"
"github.com/stretchr/testify/require"
)
func TestRpc2Header(t *testing.T) {
var header Header
require.False(t, header.Has("a"))
header.Set("a", "a")
header.Set("b", "b")
header.Set("c", "c")
require.True(t, header.Has("a"))
maxBuff := make([]byte, MaxHeaderLength+1)
maxBuff[0] = 'a'
header.Set(string(maxBuff), "x")
require.False(t, header.Has(string(maxBuff)))
header.Set("x", string(maxBuff))
require.False(t, header.Has("x"))
for idx := range [MaxHeaders]struct{}{} {
header.Set("id-"+strconv.Itoa(idx), "")
}
require.Equal(t, MaxHeaders, len(header.M))
header.Del("a")
require.False(t, header.Has("a"))
header.SetStable()
header.Set("a", "a")
require.False(t, header.Has("a"))
header.Del("b")
require.True(t, header.Has("b"))
header.Merge(header)
header = header.Clone()
header.Set("a", "a")
require.True(t, header.Has("a"))
header.Reset()
require.Nil(t, header.M)
}
func TestRpc2FixedHeader(t *testing.T) {
var header FixedHeader
require.False(t, header.Has("a"))
header.Set("a", "a")
header.Set("b", "b")
header.Set("c", "c")
require.True(t, header.Has("a"))
maxBuff := make([]byte, MaxHeaderLength+1)
maxBuff[0] = 'a'
header.Set(string(maxBuff), "x")
require.False(t, header.Has(string(maxBuff)))
header.Set("x", string(maxBuff))
require.False(t, header.Has("x"))
for idx := range [MaxHeaders]struct{}{} {
header.Set("id-"+strconv.Itoa(idx), "")
}
require.Equal(t, MaxHeaders, len(header.M))
header.Del("a")
require.False(t, header.Has("a"))
o := header.ToHeader()
header.MergeHeader(o)
header.SetStable()
header.MergeHeader(o)
header.Set("a", "a")
require.False(t, header.Has("a"))
header.Del("b")
require.True(t, header.Has("b"))
header.Reset()
require.Nil(t, header.M)
}
func TestRpc2FixedHeaderReader(t *testing.T) {
var header FixedHeader
require.Equal(t, NoBody, header.Reader())
header.Add("a", "aaaa")
b, _ := io.ReadAll(header.Reader())
require.Equal(t, header.Get("a"), string(b))
header.Set("c", "cccc")
header.SetLen("b", 8)
header.Set("b", "123456789")
b, _ = io.ReadAll(header.Reader())
require.Equal(t, "aaaa12345678cccc", string(b))
header.SetStable()
header.Set("b", "87654321")
header.Set("d", "dddd")
b, _ = io.ReadAll(header.Reader())
require.Equal(t, "aaaa87654321cccc", string(b))
r := trailerReader{
Fn: func() error { return errLimitedWrite },
Trailer: &header,
}
_, err := r.Read(nil)
require.ErrorIs(t, errLimitedWrite, err)
}

2
blobstore/common/rpc2/request.go
View File

@ -126,7 +126,7 @@ func (req *Request) request(deadline time.Time) (*Response, error) {
payloadSize += int(resp.ContentLength)
}
resp.Body = makeBodyWithTrailer(req.conn.NewSizedReader(payloadSize, frame),
req, resp.ContentLength, decode)
req, &resp.Trailer, resp.ContentLength, decode)
return resp, nil
}

102
blobstore/common/rpc2/request_test.go Normal file
View File

@ -0,0 +1,102 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"bytes"
"context"
"io"
"testing"
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
"github.com/stretchr/testify/require"
)
func handleRequestTimeout(w ResponseWriter, req *Request) error {
time.Sleep(10 * time.Second)
return w.WriteOK(nil)
}
func TestRequestTimeout(t *testing.T) {
var handler Router
handler.Register("/", handleRequestTimeout)
handler.Register("/none", handleNone)
server, cli, shutdown := newServer("tcp", &handler)
defer shutdown()
cli.RequestTimeout = 200 * time.Millisecond
req, err := NewRequest(testCtx, server.Name, "/", nil, nil)
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.ErrorIs(t, transport.ErrTimeout, err)
cli.RequestTimeout = 0
ctx, cancel := context.WithDeadline(testCtx, time.Now().Add(100*time.Millisecond))
req, err = NewRequest(ctx, server.Name, "/", nil, nil)
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.ErrorIs(t, transport.ErrTimeout, err)
cancel()
cli.Timeout = 200 * time.Millisecond
req, err = NewRequest(testCtx, server.Name, "/", nil, nil)
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.ErrorIs(t, transport.ErrTimeout, err)
cli.Timeout = 0
buff := make([]byte, 8<<20)
cli.ResponseTimeout = 200 * time.Millisecond
req, err = NewRequest(testCtx, server.Name, "/none", nil, bytes.NewReader(buff))
require.NoError(t, err)
resp, err := cli.Do(req, nil)
require.NoError(t, err)
time.Sleep(250 * time.Millisecond)
_, err = io.ReadFull(resp.Body, buff)
require.Error(t, err)
cli.ResponseTimeout = 0
cli.Timeout = time.Second
ctx, cancel = context.WithDeadline(testCtx, time.Now().Add(100*time.Millisecond))
req, err = NewRequest(ctx, server.Name, "/none", nil, bytes.NewReader(buff))
require.NoError(t, err)
resp, err = cli.Do(req, nil)
require.NoError(t, err)
time.Sleep(250 * time.Millisecond)
_, err = io.ReadFull(resp.Body, buff)
require.Error(t, err)
cancel()
}
func TestRequestErrors(t *testing.T) {
addr, cli, shutdown := newTcpServer()
defer shutdown()
cli.ConnectorConfig.Transport.MaxFrameSize = 32
req, err := NewRequest(testCtx, addr, "/", nil, nil)
require.Panics(t, func() {
req.OptionChecksum(ChecksumBlock{
Algorithm: ChecksumAlgorithm_Alg_None,
Direction: ChecksumDirection_Dir_None,
BlockSize: 1 << 10,
})
})
req.OptionCrcDownload()
req.OptionCrcDownload()
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.ErrorIs(t, ErrFrameHeader, err)
}

75
blobstore/common/rpc2/response_test.go Normal file
View File

@ -0,0 +1,75 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"testing"
"github.com/stretchr/testify/require"
)
func handleResponseDoubleOk(w ResponseWriter, req *Request) error {
w.WriteHeader(200, nil)
w.WriteHeader(200, nil)
w.WriteOK(NoParameter)
return w.WriteOK(nil)
}
func handleResponseDoubleStatus(w ResponseWriter, req *Request) error {
size := int64(10)
w.SetContentLength(size)
w.Flush()
resp := w.(*response)
resp.Write(make([]byte, size-1))
resp.Write(make([]byte, size))
resp.Write(make([]byte, size))
return nil
}
// response has wrote 200 OK
func handleResponseAfterError(w ResponseWriter, req *Request) error {
w.AfterBody(func() error { return &Error{Status: 511, Detail: "after body"} })
return w.WriteOK(nil)
}
func handleResponseClosed(w ResponseWriter, req *Request) error {
resp := w.(*response)
resp.hdr.ToString()
resp.conn.Close()
return w.WriteOK(nil)
}
func TestResponseError(t *testing.T) {
var handler Router
handler.Register("/ok", handleResponseDoubleOk)
handler.Register("/status", handleResponseDoubleStatus)
handler.Register("/after", handleResponseAfterError)
handler.Register("/closed", handleResponseClosed)
server, cli, shutdown := newServer("tcp", &handler)
defer shutdown()
req, err := NewRequest(testCtx, server.Name, "/ok", nil, nil)
require.NoError(t, err)
require.NoError(t, cli.DoWith(req, nil))
req, err = NewRequest(testCtx, server.Name, "/status", nil, nil)
require.NoError(t, err)
require.NoError(t, cli.DoWith(req, nil))
req, err = NewRequest(testCtx, server.Name, "/after", nil, nil)
require.NoError(t, err)
require.NoError(t, cli.DoWith(req, nil))
req, err = NewRequest(testCtx, server.Name, "/closed", nil, nil)
require.NoError(t, err)
require.Error(t, cli.DoWith(req, nil))
}

85
blobstore/common/rpc2/router_test.go Normal file
View File

@ -0,0 +1,85 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"fmt"
"testing"
"github.com/stretchr/testify/require"
)
func handleMiddlewareNil(w ResponseWriter, req *Request) error { return nil }
func handleMiddlewareError(w ResponseWriter, req *Request) error { return &Error{Status: 555} }
func handleRouteError(w ResponseWriter, req *Request) error { return &Error{Status: 666} }
func handleRoutePanic(w ResponseWriter, req *Request) error { panic("route") }
func TestRpc2Router(t *testing.T) {
{
var handler Router
handler.Middleware(handleMiddlewareError)
handler.Register("/", handleRouteError)
server, cli, shutdown := newServer("tcp", &handler)
defer shutdown()
req, err := NewRequest(testCtx, server.Name, "/", nil, nil)
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.Equal(t, 555, DetectStatusCode(err))
}
{
var handler Router
handler.Register("/", handleRouteError)
server, cli, shutdown := newServer("tcp", &handler)
defer shutdown()
req, err := NewRequest(testCtx, server.Name, "/404", nil, nil)
require.NoError(t, err, ErrorString(err))
err = cli.DoWith(req, nil)
require.Equal(t, 404, DetectStatusCode(err), ErrorString(err))
req, err = NewRequest(testCtx, server.Name, "/", nil, nil)
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.Equal(t, 666, DetectStatusCode(err))
}
{
var handler Router
handler.Register("/", handleRoutePanic)
server, cli, shutdown := newServer("tcp", &handler)
defer shutdown()
req, err := NewRequest(testCtx, server.Name, "/", nil, nil)
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.Equal(t, 597, DetectStatusCode(err))
handler.PanicHandler = func(ResponseWriter, *Request, interface{}, []byte) error {
return &Error{Status: 777}
}
req, err = NewRequest(testCtx, server.Name, "/", nil, nil)
require.NoError(t, err)
err = cli.DoWith(req, nil)
require.Equal(t, 777, DetectStatusCode(err), ErrorString(fmt.Errorf("panic 777")))
}
{
var handler Router
handler.Register("/", handleRoutePanic)
require.Panics(t, func() { handler.Register("/", handleRoutePanic) })
require.Panics(t, func() {
for range [1 << 10]struct{}{} {
handler.Middleware(handleMiddlewareNil)
}
handler.Middleware(handleMiddlewareNil)
})
}
}

3
blobstore/common/rpc2/rpc2.go
View File

@ -139,11 +139,10 @@ type LimitedWriter struct {
w io.Writer
a int64
n int64
e error
}
func LimitWriter(w io.Writer, limit int64) io.Writer {
return &LimitedWriter{w, limit, limit, nil}
return &LimitedWriter{w: w, a: limit, n: limit}
}
func (lw *LimitedWriter) Write(p []byte) (int, error) {

234
blobstore/common/rpc2/rpc2.pb.go
View File

@ -114,11 +114,8 @@ func (ChecksumDirection) EnumDescriptor() ([]byte, []int) {
}
type Header struct {
M map[string]string `protobuf:"bytes,1,rep,name=m,proto3" json:"m,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
M map[string]string `protobuf:"bytes,1,rep,name=m,proto3" json:"m,omitempty" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"`
}
func (m *Header) Reset() { *m = Header{} }
@ -169,11 +166,8 @@ func (m *Header) Getstable() bool {
}
type FixedValue struct {
Len uint32 `protobuf:"varint,1,opt,name=len,proto3" json:"len,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Len uint32 `protobuf:"varint,1,opt,name=len,proto3" json:"len,omitempty"`
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
}
func (m *FixedValue) Reset() { *m = FixedValue{} }
@ -224,11 +218,8 @@ func (m *FixedValue) GetValue() string {
}
type FixedHeader struct {
M map[string]FixedValue `protobuf:"bytes,1,rep,name=m,proto3" json:"m" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
M map[string]FixedValue `protobuf:"bytes,1,rep,name=m,proto3" json:"m" protobuf_key:"bytes,1,opt,name=key,proto3" protobuf_val:"bytes,2,opt,name=value,proto3"`
stable bool `protobuf:"varint,2,opt,name=stable,proto3" json:"-"`
}
func (m *FixedHeader) Reset() { *m = FixedHeader{} }
@ -279,18 +270,15 @@ func (m *FixedHeader) Getstable() bool {
}
type RequestHeader struct {
Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"`
Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"`
StreamCmd StreamCmd `protobuf:"varint,3,opt,name=stream_cmd,json=streamCmd,proto3,enum=cubefs.blobstore.common.rpc2.StreamCmd" json:"stream_cmd,omitempty"`
RemotePath string `protobuf:"bytes,4,opt,name=remote_path,json=remotePath,proto3" json:"remote_path,omitempty"`
TraceID string `protobuf:"bytes,5,opt,name=trace_id,json=traceId,proto3" json:"trace_id,omitempty"`
ContentLength int64 `protobuf:"varint,6,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"`
Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"`
Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"`
Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"`
Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"`
StreamCmd StreamCmd `protobuf:"varint,3,opt,name=stream_cmd,json=streamCmd,proto3,enum=cubefs.blobstore.common.rpc2.StreamCmd" json:"stream_cmd,omitempty"`
RemotePath string `protobuf:"bytes,4,opt,name=remote_path,json=remotePath,proto3" json:"remote_path,omitempty"`
TraceID string `protobuf:"bytes,5,opt,name=trace_id,json=traceId,proto3" json:"trace_id,omitempty"`
ContentLength int64 `protobuf:"varint,6,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"`
Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"`
Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"`
Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"`
}
func (m *RequestHeader) Reset() { *m = RequestHeader{} }
@ -390,18 +378,15 @@ func (m *RequestHeader) GetParameter() []byte {
}
type ResponseHeader struct {
Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"`
Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"`
Status int32 `protobuf:"varint,4,opt,name=status,proto3" json:"status,omitempty"`
Reason string `protobuf:"bytes,5,opt,name=reason,proto3" json:"reason,omitempty"`
Error string `protobuf:"bytes,6,opt,name=error,proto3" json:"error,omitempty"`
ContentLength int64 `protobuf:"varint,7,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"`
Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"`
Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"`
Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Version int32 `protobuf:"varint,1,opt,name=version,proto3" json:"version,omitempty"`
Magic int32 `protobuf:"varint,2,opt,name=magic,proto3" json:"magic,omitempty"`
Status int32 `protobuf:"varint,4,opt,name=status,proto3" json:"status,omitempty"`
Reason string `protobuf:"bytes,5,opt,name=reason,proto3" json:"reason,omitempty"`
Error string `protobuf:"bytes,6,opt,name=error,proto3" json:"error,omitempty"`
ContentLength int64 `protobuf:"varint,7,opt,name=content_length,json=contentLength,proto3" json:"content_length,omitempty"`
Header Header `protobuf:"bytes,8,opt,name=header,proto3" json:"header"`
Trailer FixedHeader `protobuf:"bytes,9,opt,name=trailer,proto3" json:"trailer"`
Parameter []byte `protobuf:"bytes,10,opt,name=parameter,proto3" json:"parameter,omitempty"`
}
func (m *ResponseHeader) Reset() { *m = ResponseHeader{} }
@ -501,12 +486,9 @@ func (m *ResponseHeader) GetParameter() []byte {
}
type Error struct {
Status int32 `protobuf:"varint,1,opt,name=status,proto3" json:"status,omitempty"`
Reason string `protobuf:"bytes,2,opt,name=reason,proto3" json:"reason,omitempty"`
Detail string `protobuf:"bytes,3,opt,name=detail,proto3" json:"detail,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Status int32 `protobuf:"varint,1,opt,name=status,proto3" json:"status,omitempty"`
Reason string `protobuf:"bytes,2,opt,name=reason,proto3" json:"reason,omitempty"`
Detail string `protobuf:"bytes,3,opt,name=detail,proto3" json:"detail,omitempty"`
}
func (m *Error) Reset() { *m = Error{} }
@ -564,12 +546,9 @@ func (m *Error) GetDetail() string {
}
type ChecksumBlock struct {
Algorithm ChecksumAlgorithm `protobuf:"varint,1,opt,name=algorithm,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumAlgorithm" json:"algorithm,omitempty"`
Direction ChecksumDirection `protobuf:"varint,2,opt,name=direction,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumDirection" json:"direction,omitempty"`
BlockSize uint32 `protobuf:"varint,3,opt,name=blockSize,proto3" json:"blockSize,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Algorithm ChecksumAlgorithm `protobuf:"varint,1,opt,name=algorithm,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumAlgorithm" json:"algorithm,omitempty"`
Direction ChecksumDirection `protobuf:"varint,2,opt,name=direction,proto3,enum=cubefs.blobstore.common.rpc2.ChecksumDirection" json:"direction,omitempty"`
BlockSize uint32 `protobuf:"varint,3,opt,name=blockSize,proto3" json:"blockSize,omitempty"`
}
func (m *ChecksumBlock) Reset() { *m = ChecksumBlock{} }
@ -644,54 +623,55 @@ func init() {
func init() { proto.RegisterFile("rpc2.proto", fileDescriptor_af0916bb5e6806d0) }
var fileDescriptor_af0916bb5e6806d0 = []byte{
// 745 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x55, 0x5f, 0x6f, 0xda, 0x56,
0x1c, 0xcd, 0xc5, 0xc1, 0xe0, 0x1f, 0x81, 0x79, 0x56, 0x16, 0x59, 0x51, 0x14, 0x10, 0xda, 0x16,
0x96, 0x69, 0x66, 0x62, 0x79, 0xd8, 0x1f, 0x6d, 0x52, 0x08, 0xb0, 0x20, 0x2d, 0x24, 0xba, 0x64,
0x93, 0xd6, 0x87, 0x22, 0x63, 0xdf, 0x82, 0x15, 0xdb, 0x97, 0x5e, 0x5f, 0xd2, 0xa4, 0x1f, 0xa7,
0xea, 0x87, 0x89, 0xd4, 0x97, 0x3e, 0x57, 0x15, 0xaa, 0x78, 0xec, 0xa7, 0xa8, 0xee, 0xb5, 0x1d,
0x52, 0x35, 0x45, 0xb4, 0x6f, 0x7d, 0xfb, 0x9d, 0xe3, 0x7b, 0x8e, 0x7f, 0xe7, 0xf8, 0x22, 0x00,
0xd8, 0xc4, 0x69, 0x58, 0x13, 0x46, 0x39, 0x35, 0x76, 0x9c, 0xe9, 0x90, 0x3c, 0x8a, 0xac, 0xa1,
0x4f, 0x87, 0x11, 0xa7, 0x8c, 0x58, 0x0e, 0x0d, 0x02, 0x1a, 0x5a, 0xe2, 0xcc, 0xf6, 0xe6, 0x88,
0x8e, 0xa8, 0x3c, 0x58, 0x17, 0x53, 0xac, 0xa9, 0x3e, 0x43, 0xa0, 0x1e, 0x13, 0xdb, 0x25, 0xcc,
0xf8, 0x0d, 0x50, 0x60, 0xa2, 0x8a, 0x52, 0x2b, 0x34, 0x7e, 0xb4, 0x96, 0x59, 0x59, 0xb1, 0xc0,
0x3a, 0x69, 0x87, 0x9c, 0x5d, 0x63, 0x14, 0x18, 0x7b, 0xa0, 0x46, 0xdc, 0x1e, 0xfa, 0xc4, 0xcc,
0x54, 0x50, 0x2d, 0xdf, 0xfc, 0x6a, 0x3e, 0x2b, 0x27, 0xcc, 0xdb, 0x59, 0x19, 0xfd, 0x84, 0x13,
0xb0, 0x7d, 0x00, 0x6a, 0xac, 0x32, 0x74, 0x50, 0x2e, 0xc8, 0xb5, 0x89, 0x2a, 0xa8, 0xa6, 0x61,
0x31, 0x1a, 0x9b, 0x90, 0xbd, 0xb4, 0xfd, 0x69, 0xec, 0xa1, 0xe1, 0x18, 0xfc, 0x9e, 0xf9, 0x15,
0x55, 0x0f, 0x00, 0x3a, 0xde, 0x15, 0x71, 0xff, 0x13, 0x8c, 0x50, 0xfa, 0x24, 0x94, 0xca, 0x22,
0x16, 0xe3, 0xfd, 0xca, 0xea, 0x6b, 0x04, 0x05, 0x29, 0x4b, 0xf2, 0xb5, 0x16, 0xf9, 0x7e, 0x5e,
0x9e, 0xef, 0x8e, 0x2a, 0x09, 0xd9, 0x5c, 0xbf, 0x99, 0x95, 0xd7, 0x3e, 0x29, 0xea, 0xc3, 0x25,
0x51, 0xff, 0xba, 0xbb, 0x70, 0xa1, 0x51, 0x5b, 0x61, 0x1d, 0x99, 0xfd, 0x6e, 0x29, 0xcf, 0x15,
0x28, 0x62, 0xf2, 0x78, 0x4a, 0x22, 0x9e, 0x04, 0x34, 0x21, 0x77, 0x49, 0x58, 0xe4, 0xd1, 0xb8,
0x9c, 0x2c, 0x4e, 0xa1, 0x28, 0x28, 0xb0, 0x47, 0x9e, 0x23, 0xdf, 0x97, 0xc5, 0x31, 0x30, 0x3a,
0x00, 0x11, 0x67, 0xc4, 0x0e, 0x06, 0x4e, 0xe0, 0x9a, 0x4a, 0x05, 0xd5, 0x4a, 0x8d, 0xbd, 0xe5,
0xab, 0xf4, 0xe5, 0xf9, 0xa3, 0xc0, 0xc5, 0x5a, 0x94, 0x8e, 0x46, 0x19, 0x0a, 0x8c, 0x04, 0x94,
0x93, 0xc1, 0xc4, 0xe6, 0x63, 0x73, 0x5d, 0xe6, 0x84, 0x98, 0x3a, 0xb3, 0xf9, 0xd8, 0xf8, 0x1e,
0xf2, 0x9c, 0xd9, 0x0e, 0x19, 0x78, 0xae, 0x99, 0x15, 0x4f, 0x9b, 0x85, 0xf9, 0xac, 0x9c, 0x3b,
0x17, 0x5c, 0xb7, 0x85, 0x73, 0xf2, 0x61, 0xd7, 0x35, 0xbe, 0x83, 0x92, 0x43, 0x43, 0x4e, 0x42,
0x3e, 0xf0, 0x49, 0x38, 0xe2, 0x63, 0x53, 0xad, 0xa0, 0x9a, 0x82, 0x8b, 0x09, 0xfb, 0x8f, 0x24,
0x8d, 0x26, 0xa8, 0x63, 0x99, 0xd8, 0xcc, 0xcb, 0xfa, 0xbe, 0x5d, 0xe5, 0xb6, 0x26, 0x5f, 0x30,
0x51, 0x1a, 0x5d, 0x10, 0x6f, 0xf5, 0x7c, 0xc2, 0x4c, 0x4d, 0x9a, 0xfc, 0xb0, 0xf2, 0x95, 0x48,
0x9c, 0x52, 0xbd, 0xb1, 0x03, 0xda, 0xc4, 0x66, 0x76, 0x40, 0x38, 0x61, 0x26, 0x54, 0x50, 0x6d,
0x03, 0x2f, 0x88, 0xea, 0xab, 0x0c, 0x94, 0x30, 0x89, 0x26, 0x34, 0x8c, 0xc8, 0x67, 0x7e, 0xa7,
0x2d, 0x79, 0xe5, 0xf8, 0x34, 0x92, 0xd5, 0x66, 0x71, 0x82, 0x04, 0xcf, 0x88, 0x1d, 0xd1, 0x30,
0x2e, 0x15, 0x27, 0x48, 0xb8, 0x10, 0xc6, 0x28, 0x93, 0xed, 0x69, 0x38, 0x06, 0xf7, 0x94, 0x9b,
0xfb, 0xe2, 0xcb, 0x3d, 0x85, 0x6c, 0x5b, 0x86, 0x5b, 0x54, 0x84, 0x3e, 0x52, 0x51, 0xe6, 0xbd,
0x8a, 0xb6, 0x40, 0x75, 0x09, 0xb7, 0x3d, 0x5f, 0x5e, 0x7b, 0x0d, 0x27, 0xa8, 0xfa, 0x02, 0x41,
0xf1, 0x68, 0x4c, 0x9c, 0x8b, 0x68, 0x1a, 0x34, 0x7d, 0xea, 0x5c, 0x18, 0x27, 0xa0, 0xd9, 0xfe,
0x88, 0x32, 0x8f, 0x8f, 0x03, 0x69, 0x5e, 0x6a, 0xd4, 0x97, 0xa7, 0x49, 0xf5, 0x87, 0xa9, 0x0c,
0x2f, 0x1c, 0x84, 0x9d, 0xeb, 0x31, 0xe2, 0x70, 0x2f, 0xd9, 0x69, 0x65, 0xbb, 0x56, 0x2a, 0xc3,
0x0b, 0x07, 0x51, 0xcf, 0x50, 0xac, 0xd9, 0xf7, 0x9e, 0x12, 0x19, 0xa5, 0x88, 0x17, 0xc4, 0x7e,
0x1d, 0xb4, 0xdb, 0x1f, 0xac, 0x91, 0x03, 0xa5, 0x77, 0x7a, 0xae, 0xaf, 0x89, 0xa1, 0xff, 0x7f,
0x4f, 0x47, 0x62, 0x38, 0xeb, 0x1f, 0xeb, 0x19, 0x31, 0x74, 0xba, 0x3d, 0x5d, 0xd9, 0xff, 0x13,
0xbe, 0xfe, 0x60, 0x7b, 0x63, 0x03, 0xf2, 0x87, 0xfe, 0x68, 0xd0, 0xa3, 0x21, 0xd1, 0xd7, 0x04,
0x3a, 0x62, 0xce, 0xa0, 0xdb, 0x6e, 0xb7, 0x75, 0x94, 0xa2, 0xce, 0x61, 0xff, 0x5c, 0xcf, 0xec,
0xff, 0xbd, 0x90, 0xdf, 0x6e, 0x2b, 0x8e, 0xb4, 0x3c, 0x96, 0xca, 0x01, 0xd4, 0xd6, 0x74, 0xe2,
0x93, 0x2b, 0x1d, 0x89, 0xf9, 0xdf, 0x89, 0x4f, 0x6d, 0x57, 0xcf, 0xc8, 0x53, 0xf4, 0x49, 0x28,
0x91, 0xd2, 0xfc, 0xe6, 0x66, 0xbe, 0x8b, 0x5e, 0xce, 0x77, 0xd1, 0x9b, 0xf9, 0x2e, 0x7a, 0x90,
0xb3, 0xea, 0x7f, 0x88, 0x2a, 0x86, 0xaa, 0xfc, 0xcf, 0xfa, 0xe5, 0x5d, 0x00, 0x00, 0x00, 0xff,
0xff, 0xce, 0x76, 0x64, 0x00, 0xf5, 0x06, 0x00, 0x00,
// 758 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x55, 0x5f, 0x6f, 0xf2, 0x54,
0x1c, 0xe6, 0xd0, 0x51, 0xe8, 0x8f, 0x81, 0xb5, 0x59, 0x96, 0x66, 0x59, 0x00, 0x89, 0x3a, 0x9c,
0xb1, 0x18, 0xdc, 0x85, 0x7f, 0xa2, 0xc9, 0x18, 0xe0, 0x48, 0x1c, 0x5b, 0x0e, 0xd3, 0x44, 0x2f,
0x24, 0xa5, 0x3d, 0x42, 0xb3, 0xb6, 0x07, 0x4f, 0x0f, 0x73, 0xf3, 0x53, 0xf8, 0x19, 0x8c, 0x1f,
0x66, 0x89, 0x37, 0xbb, 0x34, 0xc6, 0x10, 0xc3, 0xee, 0xfc, 0x14, 0xe6, 0x9c, 0xb6, 0x63, 0xc6,
0xbd, 0x84, 0xf7, 0xbd, 0x7b, 0xef, 0x7e, 0xcf, 0xd3, 0xf3, 0x3c, 0xfd, 0x3d, 0x4f, 0x0f, 0x01,
0x80, 0xcd, 0x9c, 0x96, 0x35, 0x63, 0x94, 0x53, 0x63, 0xdf, 0x99, 0x8f, 0xc9, 0x0f, 0x91, 0x35,
0xf6, 0xe9, 0x38, 0xe2, 0x94, 0x11, 0xcb, 0xa1, 0x41, 0x40, 0x43, 0x4b, 0x9c, 0xd9, 0xdb, 0x99,
0xd0, 0x09, 0x95, 0x07, 0x9b, 0x62, 0x8a, 0x35, 0xf5, 0x5f, 0x11, 0xa8, 0xa7, 0xc4, 0x76, 0x09,
0x33, 0x3e, 0x01, 0x14, 0x98, 0xa8, 0xa6, 0x34, 0x8a, 0xad, 0xf7, 0xad, 0x75, 0x56, 0x56, 0x2c,
0xb0, 0xce, 0xba, 0x21, 0x67, 0xb7, 0x18, 0x05, 0xc6, 0x01, 0xa8, 0x11, 0xb7, 0xc7, 0x3e, 0x31,
0xb3, 0x35, 0xd4, 0x28, 0xb4, 0xdf, 0x58, 0x2e, 0xaa, 0x09, 0xf3, 0xcf, 0xa2, 0x8a, 0x3e, 0xc0,
0x09, 0xd8, 0x3b, 0x02, 0x35, 0x56, 0x19, 0x3a, 0x28, 0x57, 0xe4, 0xd6, 0x44, 0x35, 0xd4, 0xd0,
0xb0, 0x18, 0x8d, 0x1d, 0xc8, 0x5d, 0xdb, 0xfe, 0x3c, 0xf6, 0xd0, 0x70, 0x0c, 0x3e, 0xcd, 0x7e,
0x8c, 0xea, 0x47, 0x00, 0x3d, 0xef, 0x86, 0xb8, 0xdf, 0x08, 0x46, 0x28, 0x7d, 0x12, 0x4a, 0x65,
0x09, 0x8b, 0xf1, 0x79, 0x65, 0xfd, 0x2f, 0x04, 0x45, 0x29, 0x4b, 0xf2, 0x75, 0x56, 0xf9, 0x3e,
0x5c, 0x9f, 0xef, 0x89, 0x2a, 0x09, 0xd9, 0xde, 0xba, 0x5b, 0x54, 0x33, 0x2f, 0x15, 0xf5, 0xfb,
0x35, 0x51, 0xbf, 0x78, 0xba, 0x70, 0xb1, 0xd5, 0xd8, 0x60, 0x1d, 0x99, 0xfd, 0x69, 0x29, 0xbf,
0x29, 0x50, 0xc2, 0xe4, 0xc7, 0x39, 0x89, 0x78, 0x12, 0xd0, 0x84, 0xfc, 0x35, 0x61, 0x91, 0x47,
0xe3, 0x72, 0x72, 0x38, 0x85, 0xa2, 0xa0, 0xc0, 0x9e, 0x78, 0x8e, 0x7c, 0x5f, 0x0e, 0xc7, 0xc0,
0xe8, 0x01, 0x44, 0x9c, 0x11, 0x3b, 0x18, 0x39, 0x81, 0x6b, 0x2a, 0x35, 0xd4, 0x28, 0xb7, 0x0e,
0xd6, 0xaf, 0x32, 0x94, 0xe7, 0x4f, 0x02, 0x17, 0x6b, 0x51, 0x3a, 0x1a, 0x55, 0x28, 0x32, 0x12,
0x50, 0x4e, 0x46, 0x33, 0x9b, 0x4f, 0xcd, 0x2d, 0x99, 0x13, 0x62, 0xea, 0xc2, 0xe6, 0x53, 0xe3,
0x5d, 0x28, 0x70, 0x66, 0x3b, 0x64, 0xe4, 0xb9, 0x66, 0x4e, 0x3c, 0x6d, 0x17, 0x97, 0x8b, 0x6a,
0xfe, 0x52, 0x70, 0xfd, 0x0e, 0xce, 0xcb, 0x87, 0x7d, 0xd7, 0x78, 0x07, 0xca, 0x0e, 0x0d, 0x39,
0x09, 0xf9, 0xc8, 0x27, 0xe1, 0x84, 0x4f, 0x4d, 0xb5, 0x86, 0x1a, 0x0a, 0x2e, 0x25, 0xec, 0x57,
0x92, 0x34, 0xda, 0xa0, 0x4e, 0x65, 0x62, 0xb3, 0x20, 0xeb, 0x7b, 0x7b, 0x93, 0xdb, 0x9a, 0x7c,
0xc1, 0x44, 0x69, 0xf4, 0x41, 0xbc, 0xd5, 0xf3, 0x09, 0x33, 0x35, 0x69, 0xf2, 0xde, 0xc6, 0x57,
0x22, 0x71, 0x4a, 0xf5, 0xc6, 0x3e, 0x68, 0x33, 0x9b, 0xd9, 0x01, 0xe1, 0x84, 0x99, 0x50, 0x43,
0x8d, 0x6d, 0xbc, 0x22, 0xea, 0x7f, 0x66, 0xa1, 0x8c, 0x49, 0x34, 0xa3, 0x61, 0x44, 0x5e, 0xf1,
0x3b, 0xed, 0xca, 0x2b, 0xc7, 0xe7, 0x91, 0xac, 0x36, 0x87, 0x13, 0x24, 0x78, 0x46, 0xec, 0x88,
0x86, 0x71, 0xa9, 0x38, 0x41, 0xc2, 0x85, 0x30, 0x46, 0x99, 0x6c, 0x4f, 0xc3, 0x31, 0x78, 0xa6,
0xdc, 0xfc, 0x6b, 0x5f, 0xee, 0x39, 0xe4, 0xba, 0x32, 0xdc, 0xaa, 0x22, 0xf4, 0x82, 0x8a, 0xb2,
0xff, 0xa9, 0x68, 0x17, 0x54, 0x97, 0x70, 0xdb, 0xf3, 0xe5, 0xb5, 0xd7, 0x70, 0x82, 0xea, 0xbf,
0x23, 0x28, 0x9d, 0x4c, 0x89, 0x73, 0x15, 0xcd, 0x83, 0xb6, 0x4f, 0x9d, 0x2b, 0xe3, 0x0c, 0x34,
0xdb, 0x9f, 0x50, 0xe6, 0xf1, 0x69, 0x20, 0xcd, 0xcb, 0xad, 0xe6, 0xfa, 0x34, 0xa9, 0xfe, 0x38,
0x95, 0xe1, 0x95, 0x83, 0xb0, 0x73, 0x3d, 0x46, 0x1c, 0xee, 0x25, 0x3b, 0x6d, 0x6c, 0xd7, 0x49,
0x65, 0x78, 0xe5, 0x20, 0xea, 0x19, 0x8b, 0x35, 0x87, 0xde, 0xcf, 0x44, 0x46, 0x29, 0xe1, 0x15,
0x71, 0xd8, 0x04, 0xed, 0xf1, 0x07, 0x6b, 0xe4, 0x41, 0x19, 0x9c, 0x5f, 0xea, 0x19, 0x31, 0x0c,
0xbf, 0x1d, 0xe8, 0x48, 0x0c, 0x17, 0xc3, 0x53, 0x3d, 0x2b, 0x86, 0x5e, 0x7f, 0xa0, 0x2b, 0x87,
0x9f, 0xc3, 0x9b, 0xff, 0xdb, 0xde, 0xd8, 0x86, 0xc2, 0xb1, 0x3f, 0x19, 0x0d, 0x68, 0x48, 0xf4,
0x8c, 0x40, 0x27, 0xcc, 0x19, 0xf5, 0xbb, 0xdd, 0xae, 0x8e, 0x52, 0xd4, 0x3b, 0x1e, 0x5e, 0xea,
0xd9, 0xc3, 0x2f, 0x57, 0xf2, 0xc7, 0x6d, 0xc5, 0x91, 0x8e, 0xc7, 0x52, 0x39, 0x80, 0xda, 0x99,
0xcf, 0x7c, 0x72, 0xa3, 0x23, 0x31, 0x7f, 0x3d, 0xf3, 0xa9, 0xed, 0xea, 0x59, 0x79, 0x8a, 0xfe,
0x14, 0x4a, 0xa4, 0xb4, 0xdf, 0xba, 0x5b, 0x56, 0xd0, 0xfd, 0xb2, 0x82, 0xfe, 0x5e, 0x56, 0xd0,
0x2f, 0x0f, 0x95, 0xcc, 0xfd, 0x43, 0x25, 0xf3, 0xc7, 0x43, 0x25, 0xf3, 0x5d, 0xde, 0x6a, 0x7e,
0x26, 0x6a, 0x19, 0xab, 0xf2, 0xff, 0xeb, 0xa3, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff, 0x41, 0x15,
0x6c, 0xfc, 0x01, 0x07, 0x00, 0x00,
}
func (m *Header) Marshal() (dAtA []byte, err error) {
@ -714,10 +694,6 @@ func (m *Header) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.stable {
i--
if m.stable {
@ -770,10 +746,6 @@ func (m *FixedValue) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Value) > 0 {
i -= len(m.Value)
copy(dAtA[i:], m.Value)
@ -809,10 +781,6 @@ func (m *FixedHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.stable {
i--
if m.stable {
@ -870,10 +838,6 @@ func (m *RequestHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Parameter) > 0 {
i -= len(m.Parameter)
copy(dAtA[i:], m.Parameter)
@ -958,10 +922,6 @@ func (m *ResponseHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Parameter) > 0 {
i -= len(m.Parameter)
copy(dAtA[i:], m.Parameter)
@ -1046,10 +1006,6 @@ func (m *Error) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Detail) > 0 {
i -= len(m.Detail)
copy(dAtA[i:], m.Detail)
@ -1092,10 +1048,6 @@ func (m *ChecksumBlock) MarshalToSizedBuffer(dAtA []byte) (int, error) {
_ = i
var l int
_ = l
if m.XXX_unrecognized != nil {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if m.BlockSize != 0 {
i = encodeVarintRpc2(dAtA, i, uint64(m.BlockSize))
i--
@ -1142,9 +1094,6 @@ func (m *Header) Size() (n int) {
if m.stable {
n += 2
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
@ -1161,9 +1110,6 @@ func (m *FixedValue) Size() (n int) {
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
@ -1185,9 +1131,6 @@ func (m *FixedHeader) Size() (n int) {
if m.stable {
n += 2
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
@ -1225,9 +1168,6 @@ func (m *RequestHeader) Size() (n int) {
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
@ -1265,9 +1205,6 @@ func (m *ResponseHeader) Size() (n int) {
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
@ -1288,9 +1225,6 @@ func (m *Error) Size() (n int) {
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
@ -1309,9 +1243,6 @@ func (m *ChecksumBlock) Size() (n int) {
if m.BlockSize != 0 {
n += 1 + sovRpc2(uint64(m.BlockSize))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
return n
}
@ -1509,7 +1440,6 @@ func (m *Header) Unmarshal(dAtA []byte) error {
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
@ -1611,7 +1541,6 @@ func (m *FixedValue) Unmarshal(dAtA []byte) error {
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
@ -1811,7 +1740,6 @@ func (m *FixedHeader) Unmarshal(dAtA []byte) error {
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
@ -2102,7 +2030,6 @@ func (m *RequestHeader) Unmarshal(dAtA []byte) error {
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
@ -2393,7 +2320,6 @@ func (m *ResponseHeader) Unmarshal(dAtA []byte) error {
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
@ -2527,7 +2453,6 @@ func (m *Error) Unmarshal(dAtA []byte) error {
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}
@ -2635,7 +2560,6 @@ func (m *ChecksumBlock) Unmarshal(dAtA []byte) error {
if (iNdEx + skippy) > l {
return io.ErrUnexpectedEOF
}
m.XXX_unrecognized = append(m.XXX_unrecognized, dAtA[iNdEx:iNdEx+skippy]...)
iNdEx += skippy
}
}

3
blobstore/common/rpc2/rpc2.proto
View File

@ -20,6 +20,9 @@ option go_package = "./;rpc2";
option (gogoproto.sizer_all) = true;
option (gogoproto.marshaler_all) = true;
option (gogoproto.unmarshaler_all) = true;
option (gogoproto.goproto_unkeyed_all) = false;
option (gogoproto.goproto_unrecognized_all) = false;
option (gogoproto.goproto_sizecache_all) = false;
import "gogoproto/gogo.proto";

364
blobstore/common/rpc2/rpc2_test.go Normal file
View File

@ -0,0 +1,364 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"context"
"io"
"net"
"testing"
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
_ "github.com/cubefs/cubefs/blobstore/testing/nolog"
"github.com/cubefs/cubefs/blobstore/util/log"
"github.com/cubefs/cubefs/blobstore/util/retry"
proto "github.com/gogo/protobuf/proto"
"github.com/stretchr/testify/require"
)
var (
defHandler = &Router{}
testCtx = context.Background()
)
func init() {
defHandler.Middleware(handleMiddleware1, handleMiddleware2)
defHandler.Register("/", handleNone)
defHandler.Register("/stream", handleStream)
defHandler.Register("/error", handleError)
}
func handleMiddleware1(w ResponseWriter, req *Request) error {
log.Info("handle middleware-1")
return nil
}
func handleMiddleware2(w ResponseWriter, req *Request) error {
log.Info("handle middleware-2")
return nil
}
type noCopyReadWriter struct{}
func (noCopyReadWriter) Read(p []byte) (int, error) { return len(p), nil }
func (noCopyReadWriter) Write(p []byte) (int, error) { return len(p), nil }
func handleNone(w ResponseWriter, req *Request) error {
if req.ContentLength == 0 {
req.Body.Close()
return w.WriteOK(nil)
}
req.Body.WriteTo(LimitWriter(noCopyReadWriter{}, req.ContentLength))
w.SetContentLength(req.ContentLength)
w.WriteHeader(200, nil)
_, err := w.ReadFrom(noCopyReadWriter{})
return err
}
func handleError(w ResponseWriter, req *Request) error {
return w.WriteHeader(500, nil)
}
func handleStream(_ ResponseWriter, req *Request) error {
stream := GenericServerStream[noneCodec, noneCodec]{ServerStream: req.ServerStream()}
for {
_, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
if err = stream.Send(&noneCodec{}); err != nil {
return err
}
}
}
func newTcpServer() (string, *Client, func()) {
server, client, f := newServer("tcp", defHandler)
return server.Name, client, f
}
func newServer(network string, handler Handler) (*Server, *Client, func()) {
addr := getAddress(network)
trans := transport.DefaultConfig()
trans.Version = 2
server := Server{
Name: addr,
Addresses: []NetworkAddress{{Network: network, Address: addr}},
Transport: trans,
Handler: handler,
StatDuration: 777 * time.Millisecond,
}
server.RegisterOnShutdown(func() { log.Info("shutdown") })
go func() {
if err := server.Serve(); err != nil && err != ErrServerClosed {
panic(err)
}
}()
server.waitServe()
client := Client{
ConnectorConfig: ConnectorConfig{
Transport: trans,
Network: network,
DialTimeout: 200 * time.Millisecond,
},
RetryOn: func(err error) bool {
status := DetectStatusCode(err)
return status >= 500
},
}
return &server, &client, func() {
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel()
server.Shutdown(ctx)
client.Close()
}
}
func getAddress(network string) (addr string) {
if err := retry.Timed(10, 1).On(func() error {
ln, err := net.Listen(network, "127.0.0.1:0")
if err != nil {
return err
}
if err = ln.Close(); err != nil {
return err
}
addr = ln.Addr().String()
return nil
}); err != nil {
panic(err)
}
return
}
func BenchmarkUploadDownload(b *testing.B) {
handler := &Router{}
handler.Register("/", handleNone)
server, cli, shutdown := newServer("tcp", handler)
defer shutdown()
l := int64(4 << 20)
b.SetBytes(l)
b.ResetTimer()
b.ReportAllocs()
for i := 0; i < b.N; i++ {
req, _ := NewRequest(testCtx, server.Name, "/", nil, noCopyReadWriter{})
req.ContentLength = l
resp, _ := cli.Do(req, nil)
resp.Body.WriteTo(LimitWriter(noCopyReadWriter{}, l))
resp.Body.Close()
}
}
func TestRpc2None(t *testing.T) {
{
var x noneCodec
x.Size()
x.Marshal()
x.MarshalTo(nil)
x.Unmarshal(nil)
}
{
NoBody.Read(nil)
NoBody.WriteTo(nil)
NoBody.Close()
}
{
NoParameter.Size()
NoParameter.Marshal()
NoParameter.MarshalTo(nil)
NoParameter.Unmarshal(nil)
}
{
require.Panics(t, func() {
var x codecReadWriter
x.Size()
})
}
}
func TestRpc2Pb(t *testing.T) {
{
StreamCmd_NOT.EnumDescriptor()
ChecksumAlgorithm_Alg_None.EnumDescriptor()
ChecksumDirection_Dir_None.EnumDescriptor()
}
run := func(m interface {
Codec
Reset()
String() string
ProtoMessage()
Descriptor() ([]byte, []int)
XXX_Unmarshal(b []byte) error
XXX_Marshal(b []byte, deterministic bool) ([]byte, error)
XXX_Merge(src proto.Message)
XXX_Size() int
XXX_DiscardUnknown()
}, discard bool,
) {
_ = m.String()
m.ProtoMessage()
m.Descriptor()
b, _ := m.Marshal()
m.XXX_Unmarshal(b)
m.XXX_Marshal(b, false)
m.XXX_Marshal(b, true)
m.XXX_Merge(m)
m.XXX_Size()
if !discard {
m.XXX_DiscardUnknown()
}
m.MarshalTo(b)
m.Reset()
}
{
var v *FixedValue
v.GetLen()
v.GetValue()
v = &FixedValue{Len: 1, Value: "a"}
run(v, false)
v.GetLen()
v.GetValue()
}
{
var v *Header
v.GetM()
v.Getstable()
v = &Header{}
v.Set("a", "a")
run(v, false)
v.GetM()
v.Getstable()
}
{
var v *FixedHeader
v.GetM()
v.Getstable()
v = &FixedHeader{}
v.Set("a", "a")
run(v, false)
v.GetM()
v.Getstable()
}
{
var v *RequestHeader
v.GetVersion()
v.GetMagic()
v.GetStreamCmd()
v.GetRemotePath()
v.GetTraceID()
v.GetContentLength()
v.GetHeader()
v.GetTrailer()
v.GetParameter()
v.XXX_DiscardUnknown()
v = &RequestHeader{
Version: 1,
Magic: 1,
StreamCmd: StreamCmd_NOT,
RemotePath: "/",
TraceID: "xxx",
ContentLength: 10,
Header: Header{},
Trailer: FixedHeader{},
Parameter: []byte{0xff},
}
v.Header.Set("a", "a")
v.Trailer.Set("b", "b")
run(v, true)
v.GetVersion()
v.GetMagic()
v.GetStreamCmd()
v.GetRemotePath()
v.GetTraceID()
v.GetContentLength()
v.GetHeader()
v.GetTrailer()
v.GetParameter()
}
{
var v *ResponseHeader
v.GetVersion()
v.GetMagic()
v.GetStatus()
v.GetReason()
v.GetError()
v.GetContentLength()
v.GetHeader()
v.GetTrailer()
v.GetParameter()
v.XXX_DiscardUnknown()
v = &ResponseHeader{
Version: 1,
Magic: 1,
Status: 200,
Reason: "R",
Error: "E",
ContentLength: 10,
Header: Header{},
Trailer: FixedHeader{},
Parameter: []byte{0xff},
}
v.Header.Set("a", "a")
v.Trailer.Set("b", "b")
run(v, true)
v.GetVersion()
v.GetMagic()
v.GetStatus()
v.GetReason()
v.GetError()
v.GetContentLength()
v.GetHeader()
v.GetTrailer()
v.GetParameter()
}
{
var v *Error
v.GetStatus()
v.GetReason()
v.GetDetail()
v = &Error{
Status: 100,
Reason: "R",
Detail: "E",
}
run(v, false)
v.GetStatus()
v.GetReason()
v.GetDetail()
}
{
var v *ChecksumBlock
v.GetAlgorithm()
v.GetDirection()
v.GetBlockSize()
v = &ChecksumBlock{
Algorithm: ChecksumAlgorithm_Crc_IEEE,
Direction: ChecksumDirection_Duplex,
BlockSize: 1 << 10,
}
run(v, false)
v.GetAlgorithm()
v.GetDirection()
v.GetBlockSize()
}
}

26
blobstore/common/rpc2/server.go
View File

@ -65,6 +65,7 @@ type Server struct {
StatDuration time.Duration
statOnce sync.Once
inServe atomic.Value // true when server waiting to accept
inShutdown atomic.Value // true when server is in shutdown
listenerGroup sync.WaitGroup
@ -96,6 +97,17 @@ func (s *Server) stating() {
})
}
func (s *Server) waitServe() {
for {
if val := s.inServe.Load(); val != nil {
if val.(bool) {
return
}
}
time.Sleep(10 * time.Millisecond)
}
}
func (s *Server) shuttingDown() bool {
if val := s.inShutdown.Load(); val != nil {
return val.(bool)
@ -170,12 +182,11 @@ func (s *Server) Listen(ln net.Listener) error {
s.sessions = make(map[*transport.Session]struct{})
}
_, has := s.listeners[key]
s.mu.Unlock()
if has {
s.mu.Unlock()
return nil
}
s.mu.Lock()
s.listeners[key] = struct{}{}
s.listenerGroup.Add(1)
s.mu.Unlock()
@ -194,6 +205,7 @@ func (s *Server) Listen(ln net.Listener) error {
}
for {
s.inServe.Store(true)
conn, err := ln.Accept()
if err != nil {
if s.shuttingDown() {
@ -246,7 +258,11 @@ func (s *Server) handleStream(stream *transport.Stream) {
status, reason, detail := DetectError(err)
ss.hdr.Status = int32(status)
ss.hdr.Reason = reason
ss.hdr.Error = detail.Error()
if detail != nil {
ss.hdr.Error = detail.Error()
} else {
getSpan(ctx).Warn(err)
}
} else {
ss.hdr.Status = 200
}
@ -266,6 +282,8 @@ func (s *Server) handleStream(stream *transport.Stream) {
resp.hdr.Reason = reason
if detail != nil {
resp.hdr.Error = detail.Error()
} else {
getSpan(ctx).Warn(err)
}
resp.WriteHeader(status, NoParameter)
}
@ -325,7 +343,7 @@ func (s *Server) readRequest(stream *transport.Stream) (*Request, error) {
payloadSize += int(req.ContentLength)
}
req.Body = makeBodyWithTrailer(stream.NewSizedReader(payloadSize, frame),
req, req.ContentLength, decode)
req, &req.Trailer, req.ContentLength, decode)
if hdr.StreamCmd == StreamCmd_SYN {
req.stream = &serverStream{req: req}

67
blobstore/common/rpc2/server_test.go Normal file
View File

@ -0,0 +1,67 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"context"
"testing"
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
"github.com/stretchr/testify/require"
)
func TestServerError(t *testing.T) {
{
addr := getAddress("tcp")
server := Server{
Addresses: []NetworkAddress{{Network: "network", Address: addr}},
Handler: defHandler,
}
err := server.Serve()
require.Error(t, err)
}
{
addr := getAddress("tcp")
addr1 := getAddress("tcp")
server := Server{
Addresses: []NetworkAddress{
{Network: "tcp", Address: addr},
{Network: "tcp", Address: addr1},
},
Handler: defHandler,
}
go func() { server.Serve() }()
server.waitServe()
ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
server.Shutdown(ctx)
cancel()
}
{
trans := transport.DefaultConfig()
trans.MaxFrameSize = 1 << 30
addr := getAddress("tcp")
server := Server{
Transport: trans,
Addresses: []NetworkAddress{{Network: "tcp", Address: addr}},
Handler: defHandler,
}
go func() { server.Serve() }()
server.waitServe()
connector := defaultConnector(ConnectorConfig{Network: "tcp"})
_, err := connector.Get(testCtx, addr)
require.NoError(t, err)
}
}

31
blobstore/common/rpc2/stream.go
View File

@ -17,6 +17,7 @@ package rpc2
import (
"context"
"io"
"sync"
)
type ClientStream interface {
@ -181,7 +182,7 @@ func (cs *clientStream) RecvMsg(a any) (err error) {
}()
if resp.Status > 0 { // end
cs.trailer = resp.Trailer.ToHeader()
cs.trailer.Merge(resp.Trailer.ToHeader())
cs.req.conn.Close()
if resp.Status != 200 {
return &Error{
@ -217,6 +218,7 @@ type serverStream struct {
req *Request
sentHeader bool
supplyOnce sync.Once
hdr ResponseHeader
}
@ -240,7 +242,9 @@ func (ss *serverStream) SendHeader(obj Marshaler) error {
obj = NoParameter
}
ss.sentHeader = true
ss.hdr.Status = 200
if ss.hdr.Status == 0 {
ss.hdr.Status = 200
}
ss.hdr.ContentLength = int64(obj.Size())
return ss.writeFrameMsg(&ss.hdr, obj)
}
@ -249,16 +253,14 @@ func (ss *serverStream) SetTrailer(h Header) {
ss.hdr.Trailer.MergeHeader(h)
}
func (ss *serverStream) SendMsg(a any) error {
if !ss.sentHeader {
if err := ss.SendHeader(nil); err != nil {
return err
}
}
func (ss *serverStream) SendMsg(a any) (err error) {
msg, is := a.(Codec)
if !is {
panic("rpc2: stream send message must implement rpc2.Codec")
}
if err = ss.supplyHeader(); err != nil {
return err
}
hdr := ResponseHeader{Version: Version, Magic: Magic}
hdr.ContentLength = int64(msg.Size())
@ -270,6 +272,10 @@ func (ss *serverStream) RecvMsg(a any) (err error) {
if !is {
panic("rpc2: stream recv message must implement rpc2.Codec")
}
if err = ss.supplyHeader(); err != nil {
return err
}
var req RequestHeader
frame, err := readHeaderFrame(ss.req.conn, &req)
if err != nil {
@ -292,6 +298,15 @@ func (ss *serverStream) RecvMsg(a any) (err error) {
return msg.Unmarshal(frame.Bytes(int(req.ContentLength)))
}
func (ss *serverStream) supplyHeader() (err error) {
ss.supplyOnce.Do(func() {
if !ss.sentHeader {
err = ss.SendHeader(nil)
}
})
return
}
func (ss *serverStream) writeFrameMsg(hdr *ResponseHeader, msg Marshaler) error {
size := _headerCell + hdr.Size() + msg.Size()
if size > ss.req.conn.MaxPayloadSize() {

111
blobstore/common/rpc2/stream_test.go Normal file
View File

@ -0,0 +1,111 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package rpc2
import (
"io"
"testing"
"github.com/cubefs/cubefs/blobstore/cli/common/fmt"
"github.com/stretchr/testify/require"
)
type (
streamReq = strMessage
streamResp = strMessage
)
func handleStreamFull(_ ResponseWriter, req *Request) error {
var para strMessage
req.ParseParameter(&para)
var header, trailer Header
header.Set("stream-header", "aaa")
trailer.Set("stream-trailer", "")
stream := GenericServerStream[streamReq, streamResp]{ServerStream: req.ServerStream()}
stream.Context()
stream.SetHeader(header)
stream.SetTrailer(trailer)
stream.SendHeader(&para)
defer func() {
trailer.Set("stream-trailer", "bbb")
stream.SetTrailer(trailer)
}()
for {
msg, err := stream.Recv()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
if msg.str == "error" {
return &Error{Status: 500}
}
if err = stream.Send(&streamResp{msg.str}); err != nil {
return err
}
}
}
func TestStreamBase(t *testing.T) {
handler := &Router{}
handler.Register("/", handleStreamFull)
server, cli, shutdown := newServer("tcp", handler)
defer shutdown()
sc := StreamClient[streamReq, streamResp]{Client: cli}
{
ecc := &clientStream{}
require.Panics(t, func() { ecc.RecvMsg(struct{}{}) })
require.Panics(t, func() { ecc.SendMsg(struct{}{}) })
esc := &serverStream{}
require.Panics(t, func() { esc.RecvMsg(struct{}{}) })
require.Panics(t, func() { esc.SendMsg(struct{}{}) })
}
para := strMessage{"para"}
req, err := NewStreamRequest(testCtx, server.Name, "/", &para)
require.NoError(t, err)
cc, err := sc.Streaming(req, &para)
require.NoError(t, err, cc.Context())
header, _ := cc.Header()
require.Equal(t, "aaa", header.Get("stream-header"))
trailer := cc.Trailer()
require.Equal(t, "", trailer.Get("stream-trailer"))
waitc := make(chan struct{})
go func() {
for {
_, errx := cc.Recv()
if errx != nil {
close(waitc)
return
}
}
}()
for idx := range [10]struct{}{} {
req := streamReq{str: fmt.Sprintf("request-%d", idx)}
if idx == 7 {
req.str = "error"
}
require.NoError(t, cc.Send(&req))
}
cc.CloseSend()
<-waitc
require.Equal(t, "bbb", trailer.Get("stream-trailer"))
}