forked from micro/go-micro
-
Notifications
You must be signed in to change notification settings - Fork 0
/
grpc.go
162 lines (135 loc) · 3.6 KB
/
grpc.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
// Package grpc transparently forwards the grpc protocol using a go-micro client.
package grpc
import (
"context"
"io"
"strings"
"github.com/micro/go-micro/v3/client"
"github.com/micro/go-micro/v3/client/grpc"
"github.com/micro/go-micro/v3/codec"
"github.com/micro/go-micro/v3/proxy"
"github.com/micro/go-micro/v3/server"
)
// Proxy will transparently proxy requests to the backend.
// If no backend is specified it will call a service using the client.
// If the service matches the Name it will use the server.DefaultRouter.
type Proxy struct {
// The proxy options
options proxy.Options
// Endpoint specified the fixed endpoint to call.
Endpoint string
// The client to use for outbound requests
Client client.Client
}
// read client request and write to server
func readLoop(r server.Request, s client.Stream) error {
// request to backend server
req := s.Request()
for {
// get data from client
// no need to decode it
body, err := r.Read()
if err == io.EOF {
return nil
}
if err != nil {
return err
}
// get the header from client
hdr := r.Header()
msg := &codec.Message{
Type: codec.Request,
Header: hdr,
Body: body,
}
// write the raw request
err = req.Codec().Write(msg, nil)
if err == io.EOF {
return nil
} else if err != nil {
return err
}
}
}
// ProcessMessage acts as a message exchange and forwards messages to ongoing topics
// TODO: should we look at p.Endpoint and only send to the local endpoint? probably
func (p *Proxy) ProcessMessage(ctx context.Context, msg server.Message) error {
// TODO: check that we're not broadcast storming by sending to the same topic
// that we're actually subscribed to
// directly publish to the local client
return p.Client.Publish(ctx, msg)
}
// ServeRequest honours the server.Proxy interface
func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server.Response) error {
// set default client
if p.Client == nil {
p.Client = grpc.NewClient()
}
opts := []client.CallOption{}
// service name
service := req.Service()
endpoint := req.Endpoint()
// call a specific backend
if len(p.Endpoint) > 0 {
// address:port
if parts := strings.Split(p.Endpoint, ":"); len(parts) > 1 {
opts = append(opts, client.WithAddress(p.Endpoint))
// use as service name
} else {
service = p.Endpoint
}
}
// create new request with raw bytes body
creq := p.Client.NewRequest(service, endpoint, nil, client.WithContentType(req.ContentType()))
// create new stream
stream, err := p.Client.Stream(ctx, creq, opts...)
if err != nil {
return err
}
defer stream.Close()
// create client request read loop
go readLoop(req, stream)
// get raw response
resp := stream.Response()
// create server response write loop
for {
// read backend response body
body, err := resp.Read()
if err == io.EOF {
return nil
} else if err != nil {
return err
}
// read backend response header
hdr := resp.Header()
// write raw response header to client
rsp.WriteHeader(hdr)
// write raw response body to client
err = rsp.Write(body)
if err == io.EOF {
return nil
} else if err != nil {
return err
}
}
}
func (p *Proxy) String() string {
return "grpc"
}
// NewProxy returns a new grpc proxy server
func NewProxy(opts ...proxy.Option) proxy.Proxy {
var options proxy.Options
for _, o := range opts {
o(&options)
}
p := new(Proxy)
p.Endpoint = options.Endpoint
p.Client = options.Client
return p
}
// NewSingleHostProxy returns a router which sends requests to a single backend
func NewSingleHostProxy(url string) *Proxy {
return &Proxy{
Endpoint: url,
}
}