-
Notifications
You must be signed in to change notification settings - Fork 0
/
client.go
150 lines (134 loc) · 3.75 KB
/
client.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
package client
import (
"fmt"
"os"
"github.com/rs/zerolog"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/health"
healthpb "google.golang.org/grpc/health/grpc_health_v1"
)
const (
// DefaultName prefix client name
DefaultName = "client"
)
// A Client defines parameters for making calls to an HTTP server.
// The zero value for Client is a valid configuration.
type Client struct {
cfg Config
health *health.Server
logger zerolog.Logger
}
// New create a new client
func New(opts ...Option) *Client {
return newClient(opts...)
}
// newClient will instantiate a new Client with the given config
func newClient(opts ...Option) *Client {
c := &Client{
cfg: newConfig(),
health: health.NewServer(),
}
c.logger = zerolog.New(os.Stderr).With().Timestamp().Logger()
if len(opts) > 0 {
c = c.WithOptions(opts...)
}
return c
}
// WithOptions clones the current Client, applies the supplied Options, and
// returns the resulting Client. It's safe to use concurrently.
func (c *Client) WithOptions(opts ...Option) *Client {
for _, opt := range opts {
opt.apply(c)
}
return c
}
func (c *Client) applyOptions(opts ...Option) {
for _, opt := range opts {
opt.apply(c)
}
}
// Init initialize logger and interceptors
func (c *Client) Init(opts ...Option) {
c.applyOptions(opts...)
c.logger = c.logger.With().Dict("client", zerolog.Dict().
Str("id", c.cfg.ID).
Str("name", c.cfg.Name).
Str("format", c.cfg.Format).
Str("target", c.cfg.Target),
).Logger()
c.logger.Info().Msg(fmt.Sprintf("starting %s %s, connecting to %s", c.cfg.Format, c.Name(), c.cfg.Target))
// append dial interceptor to grpc client
c.cfg.mu.Lock()
defer c.cfg.mu.Unlock()
c.cfg.UnaryClientInterceptors = append(c.cfg.UnaryClientInterceptors, dialUnaryClientInterceptor(c))
c.cfg.StreamClientInterceptors = append(c.cfg.StreamClientInterceptors, dialStreamClientInterceptor(c))
}
// Dial create a gRPC channel to communicate with the server
func (c *Client) Dial(opts ...Option) (*grpc.ClientConn, error) {
c.applyOptions(opts...)
// TODO use TLS
c.cfg.mu.Lock()
defer c.cfg.mu.Unlock()
conn, err := grpc.Dial(
c.cfg.Target,
grpc.WithInsecure(),
grpc.WithBlock(),
grpc.WithTimeout(c.cfg.Timeout),
grpc.WithUnaryInterceptor(WrapperUnaryClient(c.cfg.UnaryClientInterceptors...)),
grpc.WithStreamInterceptor(WrapperStreamClient(c.cfg.StreamClientInterceptors...)),
)
switch grpc.Code(err) {
case codes.OK:
break
default:
return nil, err
}
return conn, nil
}
// Stub to perform RPCs
func (c *Client) Stub(conn *grpc.ClientConn) interface{} {
return c.cfg.GRPCRegister(conn)
}
// Close not implemented
func (c *Client) Close() error {
return nil
}
// Name returns client name
func (c *Client) Name() string {
return c.cfg.Name
}
func (c *Client) healthRPC() {
conn, err := c.Dial()
if err != nil {
c.logger.Error().Msg(err.Error())
c.health.SetServingStatus(c.cfg.ID, healthpb.HealthCheckResponse_NOT_SERVING)
return
}
defer conn.Close()
h := healthpb.NewHealthClient(conn)
hcr, err := h.Check(context.Background(), &healthpb.HealthCheckRequest{})
if err != nil {
c.logger.Error().Msg(err.Error())
c.health.SetServingStatus(c.cfg.ID, healthpb.HealthCheckResponse_NOT_SERVING)
return
}
c.health.SetServingStatus(c.cfg.ID, hcr.Status)
}
// Health health check on client take in consideration
// a round trip to a server
func (c *Client) Health() *healthpb.HealthCheckResponse {
// perform health check RPC
c.healthRPC()
// check client status
hcr, err := c.health.Check(
context.Background(),
&healthpb.HealthCheckRequest{Service: c.cfg.ID},
)
if err != nil {
c.logger.Error().Msg(err.Error())
return &healthpb.HealthCheckResponse{Status: healthpb.HealthCheckResponse_NOT_SERVING}
}
return hcr
}