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 server stream

Browse Source 
. #22358085

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2024-08-05 17:38:30 +08:00
parent f4a5af920b
commit 2ccf906337
16 changed files with 894 additions and 264 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -28,20 +28,23 @@ type bodyAndTrailer struct {
remain int
closeOnce sync.Once
req *Request
trailer *FixedHeader
req *Request
trailerOnce sync.Once
trailer *FixedHeader
}
func (r *bodyAndTrailer) tryReadTrailer() error {
if r.remain < 0 {
panic("rpc2: response body read too much")
}
var err error
if r.remain == 0 { // try to read trailer
if _, err := r.trailer.ReadFrom(r.sr); err != nil {
return err
}
r.trailerOnce.Do(func() {
_, err = r.trailer.ReadFrom(r.sr)
})
}
return nil
return err
}
func (r *bodyAndTrailer) Read(p []byte) (int, error) {
@ -69,6 +72,7 @@ func (r *bodyAndTrailer) WriteTo(w io.Writer) (int64, error) {
}
func (r *bodyAndTrailer) Close() (err error) {
r.tryReadTrailer()
r.closeOnce.Do(func() {
err = r.sr.Close()
if r.req != nil {
@ -84,13 +88,15 @@ func (r *bodyAndTrailer) Close() (err error) {
// makeBodyWithTrailer body and trailer remain.
func makeBodyWithTrailer(sr *transport.SizedReader, l int64, req *Request, trailer *FixedHeader) Body {
return &bodyAndTrailer{
r := &bodyAndTrailer{
sr: sr,
br: io.LimitReader(sr, l),
remain: int(l),
req: req,
trailer: trailer,
}
r.tryReadTrailer()
return r
}
// readHeaderFrame try to read request or response header.
@ -109,14 +115,14 @@ func readHeaderFrame(stream *transport.Stream, hdr interface {
}()
if frame.Len() < _headerCell {
err = ErrHeaderFrame
err = ErrFrameHeader
return nil, err
}
var cell headerCell
cell.Write(frame.Bytes(_headerCell))
headerSize := cell.Get()
if frame.Len() < headerSize {
err = ErrHeaderFrame
err = ErrFrameHeader
return nil, err
}
@ -125,3 +131,15 @@ func readHeaderFrame(stream *transport.Stream, hdr interface {
}
return frame, nil
}
func writeHeaderFrame(stream *transport.Stream, hdr interface {
Size() int
MarshalToReader() io.Reader
},
) error {
var cell headerCell
cell.Set(hdr.Size())
_, err := stream.SizedWrite(io.MultiReader(cell.Reader(),
hdr.MarshalToReader()), _headerCell+hdr.Size())
return err
}

59
blobstore/common/rpc2/client.go
View File

@ -21,15 +21,11 @@ import (
"strings"
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
"github.com/cubefs/cubefs/blobstore/util/retry"
)
type Client struct {
Connector interface {
Get(ctx context.Context, addr string) (*transport.Stream, error)
Put(ctx context.Context, stream *transport.Stream) error
}
Connector Connector
Retry int
// | Request | Response Header | Response Body |
@ -40,13 +36,22 @@ type Client struct {
ResponseTimeout time.Duration
}
func (c *Client) Do(req *Request) (resp *Response, err error) {
func (c *Client) DoWith(req *Request, ret Unmarshaler) error {
resp, err := c.Do(req, ret)
if err != nil {
return err
}
resp.Body.Close()
return nil
}
func (c *Client) Do(req *Request, ret Unmarshaler) (resp *Response, err error) {
try := c.Retry
if try <= 0 {
try = 3
}
err = retry.Timed(try, 1).RuptOn(func() (bool, error) {
resp, err = c.do(req)
resp, err = c.do(req, ret)
if err != nil {
if req.Body == nil {
return true, err
@ -58,12 +63,15 @@ func (c *Client) Do(req *Request) (resp *Response, err error) {
req.Body = clientNopBody(body)
return false, err
}
return true, err
return true, nil
})
return
}
func (c *Client) do(req *Request) (*Response, error) {
func (c *Client) do(req *Request, ret Unmarshaler) (*Response, error) {
if ret == nil {
ret = NoParameter
}
conn, err := c.Connector.Get(req.Context(), req.RemoteAddr)
if err != nil {
return nil, err
@ -76,6 +84,10 @@ func (c *Client) do(req *Request) (*Response, error) {
req.conn.Close()
return nil, err
}
if err = ret.Unmarshal(resp.GetParameter()); err != nil {
resp.Body.Close()
return nil, err
}
req.conn.SetReadDeadline(c.responseDeadline(req.Context()))
return resp, nil
}
@ -102,11 +114,18 @@ func (c *Client) responseDeadline(ctx context.Context) time.Time {
return beforeContextDeadline(ctx, latestTime(timeout, respTimeout))
}
func NewRequest(ctx context.Context, addr, handler string, body io.Reader) *Request {
func NewRequest(ctx context.Context, addr, handler string, para Marshaler, body io.Reader) (*Request, error) {
rc, ok := body.(io.ReadCloser)
if !ok && body != nil {
rc = io.NopCloser(body)
}
if para == nil {
para = NoParameter
}
paraData, err := para.Marshal()
if err != nil {
return nil, err
}
req := &Request{
RequestHeader: RequestHeader{
Version: Version,
@ -114,6 +133,7 @@ func NewRequest(ctx context.Context, addr, handler string, body io.Reader) *Requ
RemoteAddr: addr,
RemoteHandler: handler,
TraceID: getSpan(ctx).TraceID(),
Parameter: paraData,
},
ctx: ctx,
Body: clientNopBody(rc),
@ -149,15 +169,15 @@ func NewRequest(ctx context.Context, addr, handler string, body io.Reader) *Requ
req.GetBody = func() (io.ReadCloser, error) { return NoBody, nil }
}
}
return req
return req, nil
}
type StreamClient[Req any, Res any] struct {
Client *Client
}
func (sc *StreamClient[Req, Res]) Streaming(req *Request) (StreamingClient[Req, Res], error) {
resp, err := sc.Client.Do(req)
func (sc *StreamClient[Req, Res]) Streaming(req *Request, ret Unmarshaler) (StreamingClient[Req, Res], error) {
resp, err := sc.Client.Do(req, ret)
if err != nil {
return nil, err
}
@ -169,7 +189,14 @@ func (sc *StreamClient[Req, Res]) Streaming(req *Request) (StreamingClient[Req,
return &GenericClientStream[Req, Res]{ClientStream: cs}, nil
}
func NewStreamRequest(ctx context.Context, addr, handler string) *Request {
func NewStreamRequest(ctx context.Context, addr, handler string, para Marshaler) (*Request, error) {
if para == nil {
para = NoParameter
}
paraData, err := para.Marshal()
if err != nil {
return nil, err
}
return &Request{
RequestHeader: RequestHeader{
Version: Version,
@ -178,9 +205,11 @@ func NewStreamRequest(ctx context.Context, addr, handler string) *Request {
StreamCmd: StreamCmd_SYN,
RemoteHandler: handler,
TraceID: getSpan(ctx).TraceID(),
Parameter: paraData,
},
ctx: ctx,
Body: NoBody,
GetBody: func() (io.ReadCloser, error) { return NoBody, nil },
AfterBody: func() error { return nil },
}
}, nil
}

68
blobstore/common/rpc2/connector.go Normal file
View File

@ -0,0 +1,68 @@
// 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"
"net"
"sync"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
)
type Dialer interface {
Dial(ctx context.Context, addr string) (net.Conn, error)
}
type Connector interface {
Get(ctx context.Context, addr string) (*transport.Stream, error)
Put(ctx context.Context, stream *transport.Stream) error
}
type sessionStreams struct {
session *transport.Session
streamC chan *transport.Stream
}
type connector struct {
dialer Dialer
config *transport.Config
mu sync.RWMutex
sessions map[string]sessionStreams
}
func DefaultConnector(dialer Dialer, config *transport.Config) Connector {
return &connector{
dialer: dialer,
config: config,
sessions: make(map[string]sessionStreams),
}
}
func (c *connector) Get(ctx context.Context, addr string) (*transport.Stream, error) {
c.mu.RLock()
ss, ok := c.sessions[addr]
c.mu.RUnlock()
if ok {
if ss.session.IsClosed() {
}
}
return nil, nil
}
func (c *connector) Put(ctx context.Context, stream *transport.Stream) error {
return nil
}

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

@ -14,10 +14,17 @@
package rpc2
import "github.com/cubefs/cubefs/blobstore/common/rpc"
import (
"errors"
"github.com/cubefs/cubefs/blobstore/common/rpc"
)
var DetectError = rpc.DetectError
var _ rpc.HTTPError = (*Error)(nil)
func (m *Error) StatusCode() int { return int(m.GetStatus()) }
func (m *Error) ErrorCode() string { return m.GetReason() }
func (m *Error) Error() string { return m.GetError_() }
func (m *Error) Error() string { return m.GetDetail() }
func (m *Error) Unwrap() error { return errors.New(m.Error()) }

27
blobstore/common/rpc2/example/client.go
View File

@ -3,6 +3,8 @@ package main
import (
"bytes"
"context"
"encoding/json"
"io"
"net"
"time"
@ -18,6 +20,21 @@ type connector struct {
func (c *connector) Get(context.Context, string) (*transport.Stream, error) { return c.stream, nil }
func (c *connector) Put(context.Context, *transport.Stream) error { return nil }
type pingPara struct {
I int
S string
}
var _ rpc2.Codec = (*pingPara)(nil)
func (p *pingPara) Marshal() ([]byte, error) {
return json.Marshal(p)
}
func (p *pingPara) Unmarshal(data []byte) error {
return json.Unmarshal(data, p)
}
func runClient() {
conn, err := net.Dial("tcp", listenon[int(time.Now().UnixNano())%len(listenon)])
if err != nil {
@ -33,16 +50,20 @@ func runClient() {
}
cli := rpc2.Client{Connector: &connector{stream: stream}}
para := pingPara{ I: 7, S: "ping string", }
buff := []byte("ping")
req := rpc2.NewRequest(context.Background(), "", "", bytes.NewReader(buff))
resp, err := cli.Do(req)
req, _ := rpc2.NewRequest(context.Background(), "", "/ping", &para, bytes.NewReader(buff))
var ret pingPara
resp, err := cli.Do(req, &ret)
if err != nil {
panic(err)
}
defer resp.Body.Close()
log.Info(resp.ResponseHeader.Status)
got, err := io.ReadAll(resp.Body)
log.Infof("got para: %+v | %v", ret, err)
log.Info("got body:", string(got), err)
time.Sleep(time.Second)
stream.Close()
}

10
blobstore/common/rpc2/example/main.go
View File

@ -8,16 +8,18 @@ import (
var listenon = []string{"localhost:9998", "localhost:9999"}
var client = flag.Bool("client", false, "is client or server")
var mode = flag.String("mode", "server", "run mode")
// main: go run main.go server.go client.go
func main() {
flag.Parse()
log.SetOutputLevel(log.Ldebug)
if *client {
runClient()
} else {
if *mode == "server" {
runServer()
} else if *mode == "client" {
runClient()
} else if *mode == "stream" {
runStream()
}
}

45
blobstore/common/rpc2/example/server.go
View File

@ -13,14 +13,55 @@ import (
type handler struct{}
func (h *handler) Handle(w rpc2.ResponseWriter, req *rpc2.Request) error {
log.Info(req.RequestHeader)
w.WriteHeader(200)
switch req.RequestHeader.RemoteHandler {
case "/ping":
return handlePing(w, req)
case "/stream":
return handleStream(w, req)
}
return nil
}
func handlePing(w rpc2.ResponseWriter, req *rpc2.Request) error {
log.Infof("%+v", req.RequestHeader)
var para pingPara
para.Unmarshal(req.GetParameter())
w.SetContentLength(req.ContentLength)
w.WriteHeader(200, &para)
buff := make([]byte, req.ContentLength)
io.ReadFull(req.Body, buff)
w.Write(buff)
return nil
}
func handleStream(_ rpc2.ResponseWriter, req *rpc2.Request) error {
var para pingPara
para.Unmarshal(req.GetParameter())
para.S = "response -> " + para.S
stream := rpc2.GenericServerStream[streamReq, streamResp]{ServerStream: req.ServerStream()}
var header, trailer rpc2.Header
header.Set("stream-header-a", "aaa")
trailer.Set("stream-trailer-b", "")
stream.SetHeader(header)
stream.SetTrailer(trailer)
stream.SendHeader(&para)
for {
req, err := stream.Recv()
if err == io.EOF {
trailer.Set("stream-trailer-b", "bbb")
stream.SetTrailer(trailer)
return nil
}
if err != nil {
return err
}
if err = stream.Send(&streamResp{"response -> " + req.str}); err != nil {
return err
}
}
}
func runServer() {
ln1, err := net.Listen("tcp", listenon[0])
if err != nil {

88
blobstore/common/rpc2/example/stream.go Normal file
View File

@ -0,0 +1,88 @@
package main
import (
"context"
"fmt"
"io"
"net"
"time"
"github.com/cubefs/cubefs/blobstore/common/rpc2"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
"github.com/cubefs/cubefs/blobstore/util/log"
)
type streamReq struct{ str string }
func (s *streamReq) Marshal() ([]byte, error) { return []byte(s.str), nil }
func (s *streamReq) Unmarshal(b []byte) error { s.str = string(b); return nil }
type streamResp struct{ str string }
func (s *streamResp) Marshal() ([]byte, error) { return []byte(s.str), nil }
func (s *streamResp) Unmarshal(b []byte) error { s.str = string(b); return nil }
var (
_ rpc2.Codec = (*streamReq)(nil)
_ rpc2.Codec = (*streamResp)(nil)
)
func runStream() {
conn, err := net.Dial("tcp", listenon[int(time.Now().UnixNano())%len(listenon)])
if err != nil {
panic(err)
}
session, err := transport.Client(conn, nil)
if err != nil {
panic(err)
}
stream, err := session.OpenStream()
if err != nil {
panic(err)
}
client := rpc2.Client{Connector: &connector{stream: stream}}
streamCli := rpc2.StreamClient[streamReq, streamResp]{Client: &client}
ctx := context.Background()
para := pingPara{I: 11, S: "stream string"}
req, err := rpc2.NewStreamRequest(ctx, "", "/stream", &para)
if err != nil {
panic(err)
}
var ret pingPara
cli, err := streamCli.Streaming(req, &ret)
if err != nil {
panic(err)
}
log.Infof("recv: para %+v", ret)
header, _ := cli.Header()
log.Infof("recv: header %+v", header.M)
log.Infof("recv: trailer %+v", cli.Trailer().M)
waitc := make(chan struct{})
go func() {
for {
resp, errx := cli.Recv()
if errx == io.EOF {
close(waitc)
return
}
if errx != nil {
panic(errx)
}
log.Info("recv:", resp.str)
}
}()
for idx := range [10]struct{}{} {
req := streamReq{str: fmt.Sprintf("request-%d", idx)}
log.Info("send:", req.str)
if err = cli.Send(&req); err != nil {
panic(err)
}
}
cli.CloseSend()
<-waitc
log.Infof("recv: trailer %+v", cli.Trailer().M)
log.Info("done.")
}

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

@ -17,8 +17,94 @@ package rpc2
import (
"bytes"
"io"
"sort"
"sync"
)
func (h *Header) newIfNil() {
if h.M == nil {
h.M = make(map[string]string)
}
}
func (h *Header) Add(key, val string) {
h.Set(key, val)
}
func (h *Header) Set(key, val string) {
h.newIfNil()
h.M[key] = val
}
func (h *Header) Del(key string) {
h.newIfNil()
delete(h.M, key)
}
func (h *Header) Get(key string) string {
h.newIfNil()
return h.M[key]
}
func (h *Header) Clone() Header {
var nh Header
nh.M = make(map[string]string, len(h.M))
for key, val := range h.M {
nh.Add(key, val)
}
return nh
}
func (h *Header) Merge(other Header) {
h.newIfNil()
for key, val := range other.M {
h.M[key] = val
}
}
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]FixedHeaderValue)
}
}
func (fh *FixedHeader) Add(key, val string) {
fh.Set(key, val)
}
func (fh *FixedHeader) Set(key, val string) {
fh.newIfNil()
if v, exist := fh.M[key]; exist {
v.Value = val
fh.M[key] = v
} else {
fh.M[key] = FixedHeaderValue{Len: int32(len(val)), Value: val}
}
}
func (fh *FixedHeader) Del(key string) {
fh.newIfNil()
delete(fh.M, key)
}
func (fh *FixedHeader) Get(key string) string {
fh.newIfNil()
return fh.M[key].Value
}
func (fh *FixedHeader) SetLen(key string, l int) {
fh.newIfNil()
fh.M[key] = FixedHeaderValue{Len: int32(l)}
}
func (fh *FixedHeader) ToHeader() Header {
h := Header{
M: make(map[string]string, len(fh.M)),
@ -29,17 +115,75 @@ func (fh *FixedHeader) ToHeader() Header {
return h
}
func (fh *FixedHeader) AllSize() int {
return 0
func (fh *FixedHeader) MergeHeader(h Header) {
for key, val := range h.M {
fh.Set(key, val)
}
}
func (fh *FixedHeader) AllSize() (n int) {
for _, v := range fh.M {
n += int(v.GetLen())
}
return
}
func (fh *FixedHeader) keys() []string {
keys := make([]string, 0, len(fh.M))
for key := range fh.M {
keys = append(keys, key)
}
sort.Strings(keys)
return keys
}
func (fh *FixedHeader) Reader() io.Reader {
return bytes.NewReader(make([]byte, 0))
if len(fh.M) == 0 {
return NoBody
}
buff := make([]byte, fh.AllSize())
off := 0
for _, key := range fh.keys() {
val := fh.M[key]
copy(buff[off:off+int(val.Len)], []byte(val.Value))
off += int(val.Len)
}
return bytes.NewReader(buff)
}
func (fh *FixedHeader) ReadFrom(r io.Reader) (int64, error) {
return 0, nil
buff := make([]byte, fh.AllSize())
n, err := io.ReadFull(r, buff)
if err != nil {
return int64(n), err
}
off := 0
for _, key := range fh.keys() {
val := fh.M[key]
val.Value = string(buff[off : off+int(val.Len)])
fh.M[key] = val
off += int(val.Len)
}
return int64(n), nil
}
// func (fh FixedHeader) Set(key, value string)
// func (fh FixedHeader) SetSize(key string, size int)
type trailerReader struct {
r io.Reader
once sync.Once
Fn func() error
Trailer FixedHeader
}
func (t *trailerReader) Read(p []byte) (n int, err error) {
t.once.Do(func() {
if t.Fn != nil {
err = t.Fn()
}
t.r = t.Trailer.Reader()
})
if err != nil {
return
}
return t.r.Read(p)
}

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

@ -49,6 +49,8 @@ type Request struct {
cli *Client
conn *transport.Stream
stream *serverStream
Body Body
GetBody func() (io.ReadCloser, error) // client side
@ -56,6 +58,10 @@ type Request struct {
AfterBody func() error
}
func (req *Request) ServerStream() ServerStream {
return req.stream
}
func (req *Request) Span() trace.Span {
return getSpan(req.ctx)
}
@ -67,7 +73,7 @@ func (req *Request) Context() context.Context {
func (req *Request) write(deadline time.Time) error {
reqHeaderSize := req.RequestHeader.Size()
if _headerCell+reqHeaderSize > req.conn.MaxPayloadSize() {
return ErrHeaderFrame
return ErrFrameHeader
}
var cell headerCell
@ -80,10 +86,7 @@ func (req *Request) write(deadline time.Time) error {
io.LimitReader(req.Body, req.ContentLength),
req.trailerReader(),
), size)
if err != nil {
return err
}
return nil
return err
}
func (req *Request) request(deadline time.Time) (*Response, error) {
@ -101,28 +104,11 @@ func (req *Request) request(deadline time.Time) (*Response, error) {
return resp, nil
}
type trailerReader struct {
once sync.Once
req *Request
r io.Reader
}
func (t *trailerReader) Read(p []byte) (n int, err error) {
t.once.Do(func() {
if fn := t.req.AfterBody; fn != nil {
if err = fn(); err != nil {
t.r = t.req.Trailer.Reader()
}
}
})
if err != nil {
return
}
return t.r.Read(p)
}
func (req *Request) trailerReader() io.Reader {
return &trailerReader{req: req}
return &trailerReader{
Fn: req.AfterBody,
Trailer: req.Trailer,
}
}
func (req *Request) WithCrc() *Request {

109
blobstore/common/rpc2/response.go
View File

@ -15,7 +15,10 @@
package rpc2
import (
"bytes"
"io"
"github.com/cubefs/cubefs/blobstore/common/rpc2/transport"
)
// server side response
@ -24,9 +27,12 @@ type ResponseWriter interface {
Header() Header
Trailer() FixedHeader
WriteHeader(status int)
WriteHeader(status int, obj Marshaler) error
Flush() error
io.Writer
io.ReaderFrom
AfterBody(func() error)
}
func (m *ResponseHeader) MarshalToReader() io.Reader {
@ -46,10 +52,21 @@ var _ ResponseWriter = &response{}
type response struct {
hdr ResponseHeader
conn *transport.Stream
connBroken bool
hasWroteHeader bool
remain int // body remain
toWrite int
toList []io.Reader
afterBody func() error
}
func (resp *response) SetContentLength(l int64) {
resp.hdr.ContentLength = l
resp.remain = int(l)
}
func (resp *response) Header() Header {
@ -60,14 +77,98 @@ func (resp *response) Trailer() FixedHeader {
return resp.hdr.Trailer
}
func (resp *response) WriteHeader(status int) {
func (resp *response) WriteHeader(status int, obj Marshaler) error {
if resp.hasWroteHeader {
return nil
}
resp.hdr.Status = int32(status)
resp.hasWroteHeader = true
if obj == nil {
obj = NoParameter
}
b, err := obj.Marshal()
if err != nil {
return err
}
resp.hdr.Parameter = b
var cell headerCell
cell.Set(resp.hdr.Size())
resp.toWrite += _headerCell + resp.hdr.Size()
resp.toList = append(resp.toList, cell.Reader(), resp.hdr.MarshalToReader())
return nil
}
func (resp *response) Write(p []byte) (int, error) {
return 0, nil
if !resp.hasWroteHeader {
if err := resp.WriteHeader(200, NoParameter); err != nil {
return 0, err
}
}
if resp.remain < len(p) {
p = p[:resp.remain]
}
resp.remain -= len(p)
resp.toWrite += len(p)
resp.toList = append(resp.toList, bytes.NewReader(p))
if resp.remain == 0 {
resp.toWrite += resp.hdr.Trailer.AllSize()
resp.toList = append(resp.toList, &trailerReader{
Fn: resp.afterBody,
Trailer: resp.hdr.Trailer,
})
}
if err := resp.Flush(); err != nil {
return 0, err
}
return len(p), nil
}
func (resp *response) ReadFrom(r io.Reader) (n int64, err error) {
return 0, nil
if !resp.hasWroteHeader {
if err := resp.WriteHeader(200, NoParameter); err != nil {
return 0, err
}
}
remain := resp.remain
resp.toWrite += remain + resp.hdr.Trailer.AllSize()
resp.toList = append(resp.toList,
io.LimitReader(r, int64(remain)),
&trailerReader{
Fn: resp.afterBody,
Trailer: resp.hdr.Trailer,
})
resp.remain = 0
if err := resp.Flush(); err != nil {
return 0, err
}
return int64(remain), nil
}
func (resp *response) Flush() error {
if len(resp.toList) == 0 {
return nil
}
_, err := resp.conn.SizedWrite(io.MultiReader(resp.toList...), resp.toWrite)
if err != nil {
resp.connBroken = true
return err
}
resp.toWrite = 0
resp.toList = resp.toList[:0]
return nil
}
func (resp *response) AfterBody(fn func() error) {
resp.afterBody = fn
}
func baseResponse() *Response {
return &Response{
ResponseHeader: ResponseHeader{
Version: Version,
Magic: Magic,
},
}
}

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

@ -37,7 +37,8 @@ var (
ErrServerClosed = errors.New("rpc2: server closed")
ErrLimitedWriter = errors.New("rpc2: request or response body wrap with LimitedWriter")
ErrHeaderFrame = errors.New("rpc2: request or response header is not in one frame")
ErrFrameHeader = errors.New("rpc2: request or response header is not in one frame")
ErrFrameProtocol = errors.New("rpc2: invalid protocol frame")
)
type Body interface {
@ -66,6 +67,13 @@ func clientNopBody(rc io.ReadCloser) Body {
return nopBody{rc}
}
var NoParameter Codec = noParameter{}
type noParameter struct{}
func (noParameter) Marshal() ([]byte, error) { return nil, nil }
func (noParameter) Unmarshal([]byte) error { return nil }
// LimitedWriter wrap Body with WriteTo
type LimitedWriter struct {
w io.Writer
@ -132,87 +140,3 @@ func latestTime(t time.Time, others ...time.Time) time.Time {
}
return t
}
// func SimpleClient() {
// var r io.Reader
// req := NewRequest(context.Background(), "localhost:9999", "/put/file", r)
// req.WithCrc()
// req.TraceID = "client-trace-id"
// req.ContentLength = 1 << 30
// req.Trailer.Set("value-before-body", "before send body")
// req.Trailer.SetSize("value-after-body", 4)
// req.AfterBody = func() error {
// req.Trailer.Set("value-after-body", "send")
// return nil
// }
// var cli *Client
// resp, _ := cli.Do(req)
// _ = resp.Status
// var w io.Writer
// resp.Body.WriteTo(w)
// resp.Body.Close()
// }
// func SimpleServer(req *Request, resp ResponseWriter) {
// _ = req.Header
// var w io.Writer
// req.Body.WriteTo(w)
// resp.SetTraceID("server-trace-id")
// resp.SetContentLength(1 << 30)
// resp.Trailer().Set("a", "b")
// resp.Trailer().SetSize("server-side", 4)
// resp.WriteHeader(200)
// var r io.Reader // 1G content
// resp.ReadFrom(r)
// resp.Trailer().Set("server-side", "server")
// resp.WriteTrailer()
// }
// type (
// reqMsg struct{}
// respMsg struct{}
// )
// func streamClieng() {
// var cli *Client
// stream, _ := cli.BidiStreaming(context.TODO(), "", "reading")
// stream.Header()
// waitc := make(chan struct{})
// go func() {
// for {
// resp, err := stream.Recv()
// if err == io.EOF {
// close(waitc)
// return
// }
// _ = resp.(respMsg)
// }
// }()
// for range [100]struct{}{} {
// stream.Send(reqMsg{})
// }
// stream.CloseSend()
// <-waitc
// stream.Trailer()
// }
// func streamServer(conn Stream, stream BidiStreamingServer) {
// stream.SetHeader(Header{})
// for {
// req, err := stream.Recv()
// if err == io.EOF {
// break
// }
// _ = req.(reqMsg)
// stream.Send(respMsg{})
// }
// stream.SetTrailer(Header{})
// }

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

@ -157,10 +157,10 @@ func (m *FixedHeaderValue) GetValue() string {
}
type FixedHeader struct {
M map[string]*FixedHeaderValue `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"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
M map[string]FixedHeaderValue `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"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *FixedHeader) Reset() { *m = FixedHeader{} }
@ -196,7 +196,7 @@ func (m *FixedHeader) XXX_DiscardUnknown() {
var xxx_messageInfo_FixedHeader proto.InternalMessageInfo
func (m *FixedHeader) GetM() map[string]*FixedHeaderValue {
func (m *FixedHeader) GetM() map[string]FixedHeaderValue {
if m != nil {
return m.M
}
@ -213,6 +213,7 @@ type RequestHeader struct {
ContentLength int64 `protobuf:"varint,7,opt,name=contentLength,proto3" json:"contentLength,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:"-"`
@ -314,6 +315,13 @@ func (m *RequestHeader) GetTrailer() FixedHeader {
return FixedHeader{}
}
func (m *RequestHeader) GetParameter() []byte {
if m != nil {
return m.Parameter
}
return nil
}
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"`
@ -323,6 +331,7 @@ type ResponseHeader struct {
ContentLength int64 `protobuf:"varint,7,opt,name=contentLength,proto3" json:"contentLength,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:"-"`
@ -417,10 +426,17 @@ func (m *ResponseHeader) GetTrailer() FixedHeader {
return FixedHeader{}
}
func (m *ResponseHeader) GetParameter() []byte {
if m != nil {
return m.Parameter
}
return nil
}
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"`
Error_ string `protobuf:"bytes,3,opt,name=error_,json=error,proto3" json:"error_,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:"-"`
@ -473,9 +489,9 @@ func (m *Error) GetReason() string {
return ""
}
func (m *Error) GetError_() string {
func (m *Error) GetDetail() string {
if m != nil {
return m.Error_
return m.Detail
}
return ""
}
@ -486,7 +502,7 @@ func init() {
proto.RegisterMapType((map[string]string)(nil), "cubefs.blobstore.common.rpc2.Header.MEntry")
proto.RegisterType((*FixedHeaderValue)(nil), "cubefs.blobstore.common.rpc2.FixedHeaderValue")
proto.RegisterType((*FixedHeader)(nil), "cubefs.blobstore.common.rpc2.FixedHeader")
proto.RegisterMapType((map[string]*FixedHeaderValue)(nil), "cubefs.blobstore.common.rpc2.FixedHeader.MEntry")
proto.RegisterMapType((map[string]FixedHeaderValue)(nil), "cubefs.blobstore.common.rpc2.FixedHeader.MEntry")
proto.RegisterType((*RequestHeader)(nil), "cubefs.blobstore.common.rpc2.RequestHeader")
proto.RegisterType((*ResponseHeader)(nil), "cubefs.blobstore.common.rpc2.ResponseHeader")
proto.RegisterType((*Error)(nil), "cubefs.blobstore.common.rpc2.Error")
@ -495,40 +511,42 @@ func init() {
func init() { proto.RegisterFile("rpc2.proto", fileDescriptor_af0916bb5e6806d0) }
var fileDescriptor_af0916bb5e6806d0 = []byte{
// 527 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x54, 0xd1, 0x8a, 0xd3, 0x40,
0x14, 0xdd, 0x49, 0x36, 0xa9, 0xbd, 0xcb, 0x2e, 0x61, 0xd8, 0x95, 0xb0, 0x48, 0x2d, 0x65, 0xc1,
0xaa, 0x90, 0x4a, 0xf5, 0x41, 0x57, 0x10, 0xac, 0xdb, 0xa5, 0x05, 0xad, 0x32, 0x15, 0x41, 0x5f,
0x24, 0x4d, 0xae, 0xdd, 0x62, 0x93, 0xa9, 0x33, 0xd3, 0xc5, 0x7e, 0x8b, 0x9f, 0xe1, 0xa3, 0x3f,
0xb0, 0x8f, 0x7e, 0x81, 0x48, 0xbf, 0x44, 0x66, 0x92, 0xd8, 0x14, 0xb4, 0x14, 0xdf, 0x7c, 0xbb,
0xe7, 0xf6, 0x9e, 0xdb, 0x73, 0xce, 0x5c, 0x02, 0x20, 0x66, 0x51, 0x3b, 0x98, 0x09, 0xae, 0x38,
0xbd, 0x11, 0xcd, 0x47, 0xf8, 0x41, 0x06, 0xa3, 0x29, 0x1f, 0x49, 0xc5, 0x05, 0x06, 0x11, 0x4f,
0x12, 0x9e, 0x06, 0x7a, 0xe6, 0xf8, 0x70, 0xcc, 0xc7, 0xdc, 0x0c, 0xb6, 0x74, 0x95, 0x71, 0x1a,
0x0b, 0x70, 0x7b, 0x18, 0xc6, 0x28, 0xe8, 0x23, 0x20, 0x89, 0x4f, 0xea, 0x76, 0x73, 0xaf, 0x7d,
0x37, 0xd8, 0xb4, 0x29, 0xc8, 0x08, 0xc1, 0x8b, 0x6e, 0xaa, 0xc4, 0x82, 0x91, 0xe4, 0xf8, 0x01,
0xb8, 0x19, 0xa0, 0x1e, 0xd8, 0x1f, 0x71, 0xe1, 0x93, 0x3a, 0x69, 0x56, 0x99, 0x2e, 0xe9, 0x21,
0x38, 0x97, 0xe1, 0x74, 0x8e, 0xbe, 0x65, 0x7a, 0x19, 0x38, 0xb5, 0x1e, 0x92, 0xc6, 0x29, 0x78,
0xe7, 0x93, 0xcf, 0x18, 0x67, 0xeb, 0xde, 0xe8, 0xbe, 0xe6, 0x4f, 0x31, 0x35, 0x7c, 0x87, 0xe9,
0xf2, 0xcf, 0xfc, 0xc6, 0x57, 0x02, 0x7b, 0x25, 0x32, 0x7d, 0xb2, 0x12, 0x7f, 0x6f, 0xb3, 0xf8,
0x12, 0xab, 0xe4, 0x20, 0xde, 0xe0, 0xe0, 0xac, 0xac, 0x60, 0xaf, 0x1d, 0x6c, 0xbd, 0xdf, 0x58,
0x2a, 0x3b, 0xfe, 0x62, 0xc3, 0x3e, 0xc3, 0x4f, 0x73, 0x94, 0x2a, 0xd7, 0xed, 0x43, 0xe5, 0x12,
0x85, 0x9c, 0xf0, 0xc2, 0x73, 0x01, 0xb5, 0xef, 0x24, 0x1c, 0x4f, 0x22, 0xf3, 0xaf, 0x0e, 0xcb,
0x00, 0xed, 0x42, 0x55, 0x2a, 0x81, 0x61, 0xf2, 0x2c, 0x89, 0x7d, 0xbb, 0x4e, 0x9a, 0x07, 0xed,
0x5b, 0x9b, 0xf5, 0x0c, 0x8b, 0x71, 0xb6, 0x62, 0xd2, 0x1a, 0x80, 0xc0, 0x84, 0x2b, 0x7c, 0x1a,
0xc7, 0xc2, 0xdf, 0x35, 0x5e, 0x4b, 0x1d, 0x7a, 0x02, 0xfb, 0x19, 0xea, 0x85, 0x69, 0x3c, 0x45,
0xe1, 0x3b, 0x66, 0x64, 0xbd, 0xa9, 0xc5, 0x2b, 0x11, 0x46, 0xd8, 0x3f, 0xf3, 0x5d, 0xf3, 0x7b,
0x01, 0x35, 0x3f, 0xe2, 0xa9, 0xc2, 0x54, 0x3d, 0xc7, 0x74, 0xac, 0x2e, 0xfc, 0x4a, 0x9d, 0x34,
0x6d, 0xb6, 0xde, 0xa4, 0x1d, 0x70, 0x2f, 0x4c, 0x0c, 0xfe, 0x35, 0x93, 0xec, 0xc9, 0x36, 0x67,
0xd7, 0xd9, 0xbd, 0xfa, 0x71, 0x73, 0x87, 0xe5, 0x4c, 0xda, 0x37, 0x1a, 0x26, 0x5a, 0x63, 0xd5,
0x2c, 0xb9, 0xbd, 0xf5, 0xf3, 0xe4, 0x9b, 0x0a, 0x7e, 0xe3, 0x9b, 0x05, 0x07, 0x0c, 0xe5, 0x8c,
0xa7, 0x12, 0xff, 0xf1, 0x79, 0xae, 0x83, 0x2b, 0x55, 0xa8, 0xe6, 0xd2, 0x64, 0xea, 0xb0, 0x1c,
0xe9, 0xbe, 0xc0, 0x50, 0xf2, 0x34, 0x0f, 0x32, 0x47, 0x7a, 0x0b, 0x0a, 0xc1, 0x45, 0x9e, 0x5f,
0x06, 0xfe, 0xcf, 0xf4, 0x06, 0xe0, 0x74, 0x8d, 0xfa, 0x55, 0x06, 0xe4, 0x2f, 0x19, 0x58, 0x6b,
0x19, 0x1c, 0x81, 0x6b, 0x6c, 0xbf, 0x37, 0xf7, 0x5c, 0x84, 0x70, 0xa7, 0x05, 0xd5, 0xdf, 0xa7,
0x4b, 0x2b, 0x60, 0x0f, 0x5e, 0xbe, 0xf6, 0x76, 0x74, 0x31, 0x7c, 0x3b, 0xf0, 0x88, 0x2e, 0x5e,
0x0d, 0x7b, 0x9e, 0xa5, 0x8b, 0xf3, 0xfe, 0xc0, 0xb3, 0x3b, 0x47, 0x57, 0xcb, 0x1a, 0xf9, 0xbe,
0xac, 0x91, 0x9f, 0xcb, 0x1a, 0x79, 0x57, 0x09, 0x5a, 0x8f, 0xb5, 0xe4, 0x91, 0x6b, 0xbe, 0x73,
0xf7, 0x7f, 0x05, 0x00, 0x00, 0xff, 0xff, 0x3e, 0xb0, 0x28, 0x77, 0x29, 0x05, 0x00, 0x00,
// 548 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xdc, 0x54, 0x51, 0x6f, 0x12, 0x4d,
0x14, 0xed, 0xb0, 0xdd, 0xe5, 0xdb, 0xdb, 0xaf, 0xcd, 0x66, 0x52, 0x9b, 0x49, 0xd3, 0x20, 0x21,
0x4d, 0x44, 0x4d, 0xb6, 0x06, 0x7d, 0xd0, 0xfa, 0x24, 0x42, 0x03, 0x89, 0x52, 0x33, 0x18, 0x13,
0x7d, 0x1b, 0x76, 0xaf, 0x94, 0xc8, 0xee, 0xe0, 0xec, 0xd0, 0xc8, 0x4f, 0xeb, 0x3f, 0xe8, 0xa3,
0xaf, 0xbe, 0x18, 0xc3, 0x2f, 0x31, 0x33, 0xbb, 0x08, 0x24, 0x4a, 0x88, 0x8f, 0xbe, 0xdd, 0x73,
0xb9, 0xe7, 0xce, 0x39, 0x67, 0x86, 0x05, 0x50, 0x93, 0xa8, 0x11, 0x4e, 0x94, 0xd4, 0x92, 0x9e,
0x44, 0xd3, 0x01, 0x7e, 0xcc, 0xc2, 0xc1, 0x58, 0x0e, 0x32, 0x2d, 0x15, 0x86, 0x91, 0x4c, 0x12,
0x99, 0x86, 0x66, 0xe6, 0xf8, 0x70, 0x28, 0x87, 0xd2, 0x0e, 0x9e, 0x99, 0x2a, 0xe7, 0xd4, 0x66,
0xe0, 0x75, 0x50, 0xc4, 0xa8, 0xe8, 0x33, 0x20, 0x09, 0x23, 0x55, 0xa7, 0xbe, 0xd7, 0x78, 0x18,
0x6e, 0xda, 0x14, 0xe6, 0x84, 0xf0, 0x75, 0x3b, 0xd5, 0x6a, 0xc6, 0x49, 0x72, 0xfc, 0x04, 0xbc,
0x1c, 0xd0, 0x00, 0x9c, 0x4f, 0x38, 0x63, 0xa4, 0x4a, 0xea, 0x3e, 0x37, 0x25, 0x3d, 0x04, 0xf7,
0x5a, 0x8c, 0xa7, 0xc8, 0x4a, 0xb6, 0x97, 0x83, 0xf3, 0xd2, 0x53, 0x52, 0x3b, 0x87, 0xe0, 0x62,
0xf4, 0x05, 0xe3, 0x7c, 0xdd, 0x3b, 0xd3, 0x37, 0xfc, 0x31, 0xa6, 0x96, 0xef, 0x72, 0x53, 0xfe,
0x9e, 0x5f, 0xbb, 0x21, 0xb0, 0xb7, 0x42, 0xa6, 0xad, 0xa5, 0xf8, 0x47, 0x9b, 0xc5, 0xaf, 0xb0,
0x0a, 0x07, 0xcd, 0xdd, 0xdb, 0xef, 0x77, 0x77, 0x8c, 0x8f, 0x78, 0x83, 0x8f, 0xd6, 0xaa, 0x8e,
0xbd, 0x46, 0xb8, 0xf5, 0x29, 0xd6, 0xd8, 0xaa, 0xef, 0x1b, 0x07, 0xf6, 0x39, 0x7e, 0x9e, 0x62,
0xa6, 0x0b, 0xf5, 0x0c, 0xca, 0xd7, 0xa8, 0xb2, 0x91, 0x5c, 0x38, 0x5f, 0x40, 0xe3, 0x3e, 0x11,
0xc3, 0x51, 0x64, 0x4f, 0x75, 0x79, 0x0e, 0x68, 0x1b, 0xfc, 0x4c, 0x2b, 0x14, 0xc9, 0xcb, 0x24,
0x66, 0x4e, 0x95, 0xd4, 0x0f, 0x1a, 0xf7, 0x36, 0xeb, 0xe9, 0x2f, 0xc6, 0xf9, 0x92, 0x49, 0x2b,
0x00, 0x0a, 0x13, 0xa9, 0xf1, 0x45, 0x1c, 0x2b, 0xb6, 0x6b, 0xbd, 0xae, 0x74, 0xe8, 0x29, 0xec,
0xe7, 0xa8, 0x23, 0xd2, 0x78, 0x8c, 0x8a, 0xb9, 0x76, 0x64, 0xbd, 0x69, 0xc4, 0x6b, 0x25, 0x22,
0xec, 0xb6, 0x98, 0x67, 0x7f, 0x5f, 0x40, 0xc3, 0x8f, 0x64, 0xaa, 0x31, 0xd5, 0xaf, 0x30, 0x1d,
0xea, 0x2b, 0x56, 0xae, 0x92, 0xba, 0xc3, 0xd7, 0x9b, 0xb4, 0x09, 0xde, 0x95, 0x8d, 0x81, 0xfd,
0x67, 0x93, 0x3d, 0xdd, 0xe6, 0xf1, 0x15, 0x77, 0x56, 0x30, 0x69, 0xd7, 0x6a, 0x18, 0x19, 0x8d,
0xbe, 0x5d, 0x72, 0x7f, 0xeb, 0xeb, 0x29, 0x36, 0x2d, 0xf8, 0xf4, 0x04, 0xfc, 0x89, 0x50, 0x22,
0x41, 0x8d, 0x8a, 0x41, 0x95, 0xd4, 0xff, 0xe7, 0xcb, 0x46, 0xed, 0x5b, 0x09, 0x0e, 0x38, 0x66,
0x13, 0x99, 0x66, 0xf8, 0x97, 0x97, 0x77, 0x04, 0x5e, 0xa6, 0x85, 0x9e, 0x66, 0x36, 0x71, 0x97,
0x17, 0xc8, 0xf4, 0x15, 0x8a, 0x4c, 0xa6, 0x45, 0xcc, 0x05, 0x32, 0x5b, 0x50, 0x29, 0xa9, 0x8a,
0x74, 0x73, 0xf0, 0x2f, 0x66, 0x7b, 0x09, 0x6e, 0xdb, 0x7a, 0x5b, 0x26, 0x44, 0xfe, 0x90, 0x50,
0x69, 0x2d, 0xa1, 0x23, 0xf0, 0x62, 0xd4, 0x62, 0x34, 0xb6, 0xff, 0x05, 0x9f, 0x17, 0xe8, 0xc1,
0x19, 0xf8, 0xbf, 0xde, 0x3d, 0x2d, 0x83, 0xd3, 0xbb, 0x7c, 0x1b, 0xec, 0x98, 0xa2, 0xff, 0xbe,
0x17, 0x10, 0x53, 0xbc, 0xe9, 0x77, 0x82, 0x92, 0x29, 0x2e, 0xba, 0xbd, 0xc0, 0x69, 0xde, 0xb9,
0x9d, 0x57, 0xc8, 0xd7, 0x79, 0x85, 0xfc, 0x98, 0x57, 0xc8, 0x87, 0x72, 0x78, 0xf6, 0xdc, 0x38,
0x1a, 0x78, 0xf6, 0x53, 0xf9, 0xf8, 0x67, 0x00, 0x00, 0x00, 0xff, 0xff, 0xa8, 0xcc, 0xbc, 0x24,
0x6c, 0x05, 0x00, 0x00,
}
func (m *Header) Marshal() (dAtA []byte, err error) {
@ -644,18 +662,16 @@ func (m *FixedHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) {
for k := range m.M {
v := m.M[k]
baseI := i
if v != nil {
{
size, err := v.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i -= size
i = encodeVarintRpc2(dAtA, i, uint64(size))
{
size, err := (&v).MarshalToSizedBuffer(dAtA[:i])
if err != nil {
return 0, err
}
i--
dAtA[i] = 0x12
i -= size
i = encodeVarintRpc2(dAtA, i, uint64(size))
}
i--
dAtA[i] = 0x12
i -= len(k)
copy(dAtA[i:], k)
i = encodeVarintRpc2(dAtA, i, uint64(len(k)))
@ -693,6 +709,13 @@ func (m *RequestHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) {
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)
i = encodeVarintRpc2(dAtA, i, uint64(len(m.Parameter)))
i--
dAtA[i] = 0x52
}
{
size, err := m.Trailer.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
@ -781,6 +804,13 @@ func (m *ResponseHeader) MarshalToSizedBuffer(dAtA []byte) (int, error) {
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)
i = encodeVarintRpc2(dAtA, i, uint64(len(m.Parameter)))
i--
dAtA[i] = 0x52
}
{
size, err := m.Trailer.MarshalToSizedBuffer(dAtA[:i])
if err != nil {
@ -862,10 +892,10 @@ func (m *Error) MarshalToSizedBuffer(dAtA []byte) (int, error) {
i -= len(m.XXX_unrecognized)
copy(dAtA[i:], m.XXX_unrecognized)
}
if len(m.Error_) > 0 {
i -= len(m.Error_)
copy(dAtA[i:], m.Error_)
i = encodeVarintRpc2(dAtA, i, uint64(len(m.Error_)))
if len(m.Detail) > 0 {
i -= len(m.Detail)
copy(dAtA[i:], m.Detail)
i = encodeVarintRpc2(dAtA, i, uint64(len(m.Detail)))
i--
dAtA[i] = 0x1a
}
@ -944,12 +974,8 @@ func (m *FixedHeader) Size() (n int) {
for k, v := range m.M {
_ = k
_ = v
l = 0
if v != nil {
l = v.Size()
l += 1 + sovRpc2(uint64(l))
}
mapEntrySize := 1 + len(k) + sovRpc2(uint64(len(k))) + l
l = v.Size()
mapEntrySize := 1 + len(k) + sovRpc2(uint64(len(k))) + 1 + l + sovRpc2(uint64(l))
n += mapEntrySize + 1 + sovRpc2(uint64(mapEntrySize))
}
}
@ -993,6 +1019,10 @@ func (m *RequestHeader) Size() (n int) {
n += 1 + l + sovRpc2(uint64(l))
l = m.Trailer.Size()
n += 1 + l + sovRpc2(uint64(l))
l = len(m.Parameter)
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
@ -1029,6 +1059,10 @@ func (m *ResponseHeader) Size() (n int) {
n += 1 + l + sovRpc2(uint64(l))
l = m.Trailer.Size()
n += 1 + l + sovRpc2(uint64(l))
l = len(m.Parameter)
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
if m.XXX_unrecognized != nil {
n += len(m.XXX_unrecognized)
}
@ -1048,7 +1082,7 @@ func (m *Error) Size() (n int) {
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
l = len(m.Error_)
l = len(m.Detail)
if l > 0 {
n += 1 + l + sovRpc2(uint64(l))
}
@ -1403,10 +1437,10 @@ func (m *FixedHeader) Unmarshal(dAtA []byte) error {
return io.ErrUnexpectedEOF
}
if m.M == nil {
m.M = make(map[string]*FixedHeaderValue)
m.M = make(map[string]FixedHeaderValue)
}
var mapkey string
var mapvalue *FixedHeaderValue
mapvalue := &FixedHeaderValue{}
for iNdEx < postIndex {
entryPreIndex := iNdEx
var wire uint64
@ -1500,7 +1534,7 @@ func (m *FixedHeader) Unmarshal(dAtA []byte) error {
iNdEx += skippy
}
}
m.M[mapkey] = mapvalue
m.M[mapkey] = *mapvalue
iNdEx = postIndex
default:
iNdEx = preIndex
@ -1791,6 +1825,40 @@ func (m *RequestHeader) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 10:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Parameter", wireType)
}
var byteLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRpc2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
byteLen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if byteLen < 0 {
return ErrInvalidLengthRpc2
}
postIndex := iNdEx + byteLen
if postIndex < 0 {
return ErrInvalidLengthRpc2
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Parameter = append(m.Parameter[:0], dAtA[iNdEx:postIndex]...)
if m.Parameter == nil {
m.Parameter = []byte{}
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipRpc2(dAtA[iNdEx:])
@ -2048,6 +2116,40 @@ func (m *ResponseHeader) Unmarshal(dAtA []byte) error {
return err
}
iNdEx = postIndex
case 10:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Parameter", wireType)
}
var byteLen int
for shift := uint(0); ; shift += 7 {
if shift >= 64 {
return ErrIntOverflowRpc2
}
if iNdEx >= l {
return io.ErrUnexpectedEOF
}
b := dAtA[iNdEx]
iNdEx++
byteLen |= int(b&0x7F) << shift
if b < 0x80 {
break
}
}
if byteLen < 0 {
return ErrInvalidLengthRpc2
}
postIndex := iNdEx + byteLen
if postIndex < 0 {
return ErrInvalidLengthRpc2
}
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Parameter = append(m.Parameter[:0], dAtA[iNdEx:postIndex]...)
if m.Parameter == nil {
m.Parameter = []byte{}
}
iNdEx = postIndex
default:
iNdEx = preIndex
skippy, err := skipRpc2(dAtA[iNdEx:])
@ -2152,7 +2254,7 @@ func (m *Error) Unmarshal(dAtA []byte) error {
iNdEx = postIndex
case 3:
if wireType != 2 {
return fmt.Errorf("proto: wrong wireType = %d for field Error_", wireType)
return fmt.Errorf("proto: wrong wireType = %d for field Detail", wireType)
}
var stringLen uint64
for shift := uint(0); ; shift += 7 {
@ -2180,7 +2282,7 @@ func (m *Error) Unmarshal(dAtA []byte) error {
if postIndex > l {
return io.ErrUnexpectedEOF
}
m.Error_ = string(dAtA[iNdEx:postIndex])
m.Detail = string(dAtA[iNdEx:postIndex])
iNdEx = postIndex
default:
iNdEx = preIndex

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

@ -33,7 +33,7 @@ message FixedHeaderValue {
}
message FixedHeader {
map<string, FixedHeaderValue> m = 1;
map<string, FixedHeaderValue> m = 1 [(gogoproto.nullable) = false];
}
enum StreamCmd {
@ -55,6 +55,8 @@ message RequestHeader {
Header header = 8 [(gogoproto.nullable) = false];
FixedHeader trailer = 9 [(gogoproto.nullable) = false];
bytes parameter = 10;
}
message ResponseHeader {
@ -68,10 +70,12 @@ message ResponseHeader {
Header header = 8 [(gogoproto.nullable) = false];
FixedHeader trailer = 9 [(gogoproto.nullable) = false];
bytes parameter = 10;
}
message Error {
int32 status = 1;
string reason = 2;
string error_ = 3;
string detail = 3;
}

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

@ -16,7 +16,7 @@ package rpc2
import (
"context"
"io"
"errors"
"net"
"sync"
"sync/atomic"
@ -205,31 +205,39 @@ func (s *Server) handleStream(stream *transport.Stream) {
}
ctx = req.Context()
resp := &response{}
if ss := req.stream; ss != nil {
if err = s.Handler.Handle(nil, req); err != nil {
status, reason, detail := DetectError(err)
ss.hdr.Status = int32(status)
ss.hdr.Reason = reason
ss.hdr.Error = detail.Error()
} else {
ss.hdr.Status = 200
}
ss.SendHeader(NoParameter)
stream.Close()
return err
}
resp := &response{conn: stream}
if err = s.Handler.Handle(resp, req); err != nil {
return err
status, reason, detail := DetectError(err)
resp.hdr.Reason = reason
resp.hdr.Error = detail.Error()
resp.WriteHeader(status, NoParameter)
}
respHeaderSize := resp.hdr.Size()
if _headerCell+respHeaderSize > stream.MaxPayloadSize() {
return ErrHeaderFrame
}
var cell headerCell
cell.Set(respHeaderSize)
size := _headerCell + respHeaderSize + int(resp.hdr.ContentLength) + resp.hdr.Trailer.AllSize()
_, err = stream.SizedWrite(io.MultiReader(cell.Reader(),
resp.hdr.MarshalToReader(),
// io.LimitReader(req.Body, req.ContentLength),
req.trailerReader(),
), size)
if err != nil {
if err = resp.Flush(); err != nil {
return err
}
if !req.Body.(*bodyAndTrailer).sr.Finished() || resp.connBroken {
return errors.New("stream conn has broken")
}
}
}(); err != nil {
span := trace.SpanFromContextSafe(ctx)
span.Errorf("stream(%d, %v, %v) %s", stream.ID(), stream.LocalAddr(), stream.RemoteAddr(), err.Error())
stream.Close()
}
}
@ -240,16 +248,28 @@ func (s *Server) readRequest(stream *transport.Stream) (*Request, error) {
return nil, err
}
switch hdr.StreamCmd {
case StreamCmd_NOT, StreamCmd_SYN:
case StreamCmd_PSH, StreamCmd_FIN:
return nil, ErrFrameProtocol
default:
return nil, ErrFrameProtocol
}
traceID := hdr.TraceID
if traceID == "" {
traceID = trace.RandomID().String()
}
_, ctx := trace.StartSpanFromContextWithTraceID(context.Background(), "", traceID)
req := &Request{RequestHeader: hdr, ctx: ctx}
req := &Request{RequestHeader: hdr, ctx: ctx, conn: stream}
req.Body = makeBodyWithTrailer(
stream.NewSizedReader(int(req.ContentLength)+req.Trailer.AllSize(), frame),
req.ContentLength, nil, &req.Trailer)
if hdr.StreamCmd == StreamCmd_SYN {
req.stream = &serverStream{req: req}
}
return req, nil
}

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

@ -15,16 +15,23 @@
package rpc2
import (
"bytes"
"context"
"io"
)
type Codec interface {
type Marshaler interface {
Marshal() ([]byte, error)
}
type Unmarshaler interface {
Unmarshal([]byte) error
}
type Codec interface {
Marshaler
Unmarshaler
}
type ClientStream interface {
Context() context.Context
@ -41,7 +48,7 @@ type ServerStream interface {
Context() context.Context
SetHeader(Header) error
SendHeader(Header) error
SendHeader(obj Marshaler) error
SetTrailer(Header)
SendMsg(a any) error
@ -156,6 +163,7 @@ func (cs *clientStream) Trailer() Header {
func (cs *clientStream) CloseSend() error {
req := cs.newRequest()
req.StreamCmd = StreamCmd_FIN
req.Trailer = cs.trailer.ToFixedHeader()
return req.write(req.cli.requestDeadline(req.ctx))
}
@ -171,8 +179,8 @@ func (cs *clientStream) SendMsg(a any) error {
req := cs.newRequest()
req.StreamCmd = StreamCmd_PSH
req.ContentLength = int64(len(b))
req.Body = clientNopBody(io.NopCloser(bytes.NewReader(b)))
req.Parameter = b
req.Body = NoBody
return req.write(req.cli.requestDeadline(req.ctx))
}
@ -183,8 +191,8 @@ func (cs *clientStream) RecvMsg(a any) (err error) {
}
conn := cs.req.conn
resp := &Response{}
frame, err := readHeaderFrame(conn, &resp.ResponseHeader)
var resp ResponseHeader
frame, err := readHeaderFrame(conn, &resp)
if err != nil {
return
}
@ -194,28 +202,20 @@ func (cs *clientStream) RecvMsg(a any) (err error) {
}
}()
if resp.Status > 0 { // end of
if resp.Status > 0 { // end
cs.trailer = resp.Trailer.ToHeader()
cs.req.conn.Close()
if resp.Status != 200 {
return &Error{
Status: resp.Status,
Reason: resp.Reason,
Error_: resp.Error,
Detail: resp.Error,
}
}
return nil
return io.EOF
}
// TODO: unmarshal from more frame
if frame.Len() != int(resp.ContentLength) {
err = ErrHeaderFrame
return
}
if err = msg.Unmarshal(frame.Bytes(int(resp.ContentLength))); err != nil {
frame.Close()
return
}
err = msg.Unmarshal(resp.Parameter)
return
}
@ -226,3 +226,78 @@ func (cs *clientStream) newRequest() *Request {
req.conn = cs.req.conn
return req
}
type serverStream struct {
req *Request
hdr ResponseHeader
}
var _ ServerStream = (*serverStream)(nil)
func (ss *serverStream) Context() context.Context {
return ss.req.Context()
}
func (ss *serverStream) SetHeader(h Header) error {
ss.hdr.Header.Merge(h)
return nil
}
func (ss *serverStream) SendHeader(obj Marshaler) error {
if obj == nil {
obj = NoParameter
}
b, err := obj.Marshal()
if err != nil {
return err
}
ss.hdr.Parameter = b
return writeHeaderFrame(ss.req.conn, &ss.hdr)
}
func (ss *serverStream) SetTrailer(h Header) {
ss.hdr.Trailer.MergeHeader(h)
}
func (ss *serverStream) SendMsg(a any) error {
msg, is := a.(Codec)
if !is {
panic("rpc2: stream send message must implement rpc2.Codec")
}
b, err := msg.Marshal()
if err != nil {
return err
}
resp := baseResponse()
hdr := resp.ResponseHeader
hdr.Parameter = b
return writeHeaderFrame(ss.req.conn, &hdr)
}
func (ss *serverStream) RecvMsg(a any) (err error) {
msg, is := a.(Codec)
if !is {
panic("rpc2: stream recv message must implement rpc2.Codec")
}
_ = msg
var req RequestHeader
frame, err := readHeaderFrame(ss.req.conn, &req)
if err != nil {
return
}
defer func() {
if errClose := frame.Close(); err == nil {
err = errClose
}
}()
if req.StreamCmd == StreamCmd_FIN {
err = io.EOF
return
}
err = msg.Unmarshal(req.GetParameter())
return
}