1
0
mirror of https://github.com/go-micro/go-micro.git synced 2024-12-24 10:07:04 +02:00

Use client.Call for non streaming requests

This commit is contained in:
Asim Aslam 2019-08-16 17:24:17 +01:00
parent 0b0eee41d0
commit 4495ca3839
4 changed files with 23 additions and 22 deletions

View File

@ -59,9 +59,6 @@ type CallOptions struct {
// Middleware for low level call func
CallWrappers []CallWrapper
// SendEOS specifies whether to send EOS
SendEOS bool
// Other options for implementations of the interface
// can be stored in a context
Context context.Context
@ -308,13 +305,6 @@ func WithDialTimeout(d time.Duration) CallOption {
}
}
// SendEOS specifies whether to send the end of stream message
func SendEOS(b bool) CallOption {
return func(o *CallOptions) {
o.SendEOS = b
}
}
// Request Options
func WithContentType(ct string) RequestOption {

View File

@ -118,7 +118,7 @@ func (r *rpcClient) call(ctx context.Context, node *registry.Node, req Request,
codec: codec,
closed: make(chan bool),
release: func(err error) { r.pool.Release(c, err) },
sendEOS: opts.SendEOS,
sendEOS: false,
}
// close the stream on exiting this function
defer stream.Close()
@ -244,7 +244,7 @@ func (r *rpcClient) stream(ctx context.Context, node *registry.Node, req Request
// used to close the stream
closed: make(chan bool),
// signal the end of stream,
sendEOS: opts.SendEOS,
sendEOS: true,
// release func
release: func(err error) { r.pool.Release(c, err) },
}

View File

@ -249,6 +249,12 @@ func (c *rpcCodec) ReadHeader(m *codec.Message, r codec.MessageType) error {
func (c *rpcCodec) ReadBody(b interface{}) error {
// read body
// read raw data
if v, ok := b.(*raw.Frame); ok {
v.Data = c.buf.rbuf.Bytes()
return nil
}
if err := c.codec.ReadBody(b); err != nil {
return errors.InternalServerError("go.micro.client.codec", err.Error())
}

View File

@ -220,9 +220,21 @@ func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server
// create new request with raw bytes body
creq := p.Client.NewRequest(service, endpoint, &bytes.Frame{body}, client.WithContentType(req.ContentType()))
// not a stream so make a client.Call request
if !req.Stream() {
// specify not to send eos
opts = append(opts, client.SendEOS(false))
crsp := new(bytes.Frame)
// make a call to the backend
if err := p.Client.Call(ctx, creq, crsp, opts...); err != nil {
return err
}
// write the response
if err := rsp.Write(crsp.Data); err != nil {
return err
}
return nil
}
// create new stream
@ -233,9 +245,7 @@ func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server
defer stream.Close()
// create client request read loop if streaming
if req.Stream() {
go readLoop(req, stream)
}
go readLoop(req, stream)
// get raw response
resp := stream.Response()
@ -273,11 +283,6 @@ func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server
} else if err != nil {
return err
}
// we don't continue unless its a stream
if !req.Stream() {
return nil
}
}
}