Skip to content

Commit 33d6278

Browse files
janezpodhostnikjanezpodhostnik
committed
Use flow-go Components for composing
1 parent 1d626c7 commit 33d6278

File tree

14 files changed

+2555
-1410
lines changed

14 files changed

+2555
-1410
lines changed

‎api/profiler.go

+60-20
Original file line numberDiff line numberDiff line change
@@ -7,54 +7,94 @@ import (
77
"net/http"
88
_ "net/http/pprof"
99
"strconv"
10+
"time"
11+
12+
"github.com/onflow/flow-go/module/component"
13+
"github.com/onflow/flow-go/module/irrecoverable"
1014

1115
"github.com/rs/zerolog"
1216
)
1317

1418
type ProfileServer struct {
15-
logger zerolog.Logger
19+
log zerolog.Logger
1620
server *http.Server
1721
endpoint string
22+
23+
startupCompleted chan struct{}
1824
}
1925

26+
var _ component.Component = (*ProfileServer)(nil)
27+
2028
func NewProfileServer(
2129
logger zerolog.Logger,
2230
host string,
2331
port int,
2432
) *ProfileServer {
2533
endpoint := net.JoinHostPort(host, strconv.Itoa(port))
2634
return &ProfileServer{
27-
logger: logger,
28-
server: &http.Server{Addr: endpoint},
29-
endpoint: endpoint,
35+
log: logger,
36+
server: &http.Server{Addr: endpoint},
37+
endpoint: endpoint,
38+
startupCompleted: make(chan struct{}),
3039
}
3140
}
3241

33-
func (s *ProfileServer) ListenAddr() string {
34-
return s.endpoint
35-
}
42+
func (s *ProfileServer) Start(ctx irrecoverable.SignalerContext) {
43+
defer close(s.startupCompleted)
44+
45+
s.server.BaseContext = func(_ net.Listener) context.Context {
46+
return ctx
47+
}
3648

37-
func (s *ProfileServer) Start() {
3849
go func() {
39-
err := s.server.ListenAndServe()
40-
if err != nil {
50+
s.log.Info().Msgf("Profiler server started: %s", s.endpoint)
51+
52+
if err := s.server.ListenAndServe(); err != nil {
53+
// http.ErrServerClosed is returned when Close or Shutdown is called
54+
// we don't consider this an error, so print this with debug level instead
4155
if errors.Is(err, http.ErrServerClosed) {
42-
s.logger.Warn().Msg("Profiler server shutdown")
43-
return
56+
s.log.Debug().Err(err).Msg("Profiler server shutdown")
57+
} else {
58+
s.log.Err(err).Msg("error running profiler server")
4459
}
45-
s.logger.Err(err).Msg("failed to start Profiler server")
46-
panic(err)
4760
}
4861
}()
4962
}
5063

51-
func (s *ProfileServer) Stop() error {
52-
ctx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
53-
defer cancel()
64+
func (s *ProfileServer) Ready() <-chan struct{} {
65+
ready := make(chan struct{})
66+
67+
go func() {
68+
<-s.startupCompleted
69+
close(ready)
70+
}()
5471

55-
return s.server.Shutdown(ctx)
72+
return ready
5673
}
5774

58-
func (s *ProfileServer) Close() error {
59-
return s.server.Close()
75+
func (s *ProfileServer) Done() <-chan struct{} {
76+
done := make(chan struct{})
77+
go func() {
78+
<-s.startupCompleted
79+
defer close(done)
80+
81+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
82+
defer cancel()
83+
84+
err := s.server.Shutdown(ctx)
85+
if err == nil {
86+
s.log.Info().Msg("Profiler server graceful shutdown completed")
87+
}
88+
89+
if errors.Is(err, ctx.Err()) {
90+
s.log.Warn().Msg("Profiler server graceful shutdown timed out")
91+
err := s.server.Close()
92+
if err != nil {
93+
s.log.Err(err).Msg("error closing profiler server")
94+
}
95+
} else {
96+
s.log.Err(err).Msg("error shutting down profiler server")
97+
}
98+
}()
99+
return done
60100
}

‎api/server.go

+72-40
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,9 @@ import (
1818
"strings"
1919
"time"
2020

21+
"github.com/onflow/flow-go/module/component"
22+
"github.com/onflow/flow-go/module/irrecoverable"
23+
2124
gethVM "github.com/onflow/go-ethereum/core/vm"
2225
gethLog "github.com/onflow/go-ethereum/log"
2326
"github.com/onflow/go-ethereum/rpc"
@@ -55,8 +58,12 @@ type Server struct {
5558

5659
config config.Config
5760
collector metrics.Collector
61+
62+
startupCompleted chan struct{}
5863
}
5964

65+
var _ component.Component = (*Server)(nil)
66+
6067
const (
6168
shutdownTimeout = 5 * time.Second
6269
batchRequestLimit = 50
@@ -77,10 +84,11 @@ func NewServer(
7784
gethLog.SetDefault(gethLog.NewLogger(zeroSlog))
7885

7986
return &Server{
80-
logger: logger,
81-
timeouts: rpc.DefaultHTTPTimeouts,
82-
config: cfg,
83-
collector: collector,
87+
logger: logger,
88+
timeouts: rpc.DefaultHTTPTimeouts,
89+
config: cfg,
90+
collector: collector,
91+
startupCompleted: make(chan struct{}),
8492
}
8593
}
8694

@@ -177,9 +185,10 @@ func (h *Server) disableWS() bool {
177185
}
178186

179187
// Start starts the HTTP server if it is enabled and not already running.
180-
func (h *Server) Start() error {
188+
func (h *Server) Start(ctx irrecoverable.SignalerContext) {
189+
defer close(h.startupCompleted)
181190
if h.endpoint == "" || h.listener != nil {
182-
return nil // already running or not configured
191+
return // already running or not configured
183192
}
184193

185194
// Initialize the server.
@@ -190,16 +199,21 @@ func (h *Server) Start() error {
190199
h.server.ReadHeaderTimeout = h.timeouts.ReadHeaderTimeout
191200
h.server.WriteTimeout = h.timeouts.WriteTimeout
192201
h.server.IdleTimeout = h.timeouts.IdleTimeout
202+
h.server.BaseContext = func(_ net.Listener) context.Context {
203+
return ctx
204+
}
193205
}
194206

207+
listenConfig := net.ListenConfig{}
195208
// Start the server.
196-
listener, err := net.Listen("tcp", h.endpoint)
209+
listener, err := listenConfig.Listen(ctx, "tcp", h.endpoint)
197210
if err != nil {
198211
// If the server fails to start, we need to clear out the RPC and WS
199212
// configurations so they can be configured another time.
200213
h.disableRPC()
201214
h.disableWS()
202-
return err
215+
ctx.Throw(err)
216+
return
203217
}
204218

205219
h.listener = listener
@@ -211,7 +225,7 @@ func (h *Server) Start() error {
211225
return
212226
}
213227
h.logger.Err(err).Msg("failed to start API server")
214-
panic(err)
228+
ctx.Throw(err)
215229
}
216230
}()
217231

@@ -223,8 +237,17 @@ func (h *Server) Start() error {
223237
url := fmt.Sprintf("ws://%v", listener.Addr())
224238
h.logger.Info().Msgf("JSON-RPC over WebSocket enabled: %s", url)
225239
}
240+
}
226241

227-
return nil
242+
func (h *Server) Ready() <-chan struct{} {
243+
ready := make(chan struct{})
244+
245+
go func() {
246+
<-h.startupCompleted
247+
close(ready)
248+
}()
249+
250+
return ready
228251
}
229252

230253
// disableRPC stops the JSON-RPC over HTTP handler.
@@ -294,41 +317,50 @@ func (h *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
294317
w.WriteHeader(http.StatusNotFound)
295318
}
296319

297-
// Stop shuts down the HTTP server.
298-
func (h *Server) Stop() {
299-
if h.listener == nil {
300-
return // not running
301-
}
320+
// Done shuts down the HTTP server.
321+
func (h *Server) Done() <-chan struct{} {
322+
done := make(chan struct{})
302323

303-
// Shut down the server.
304-
httpHandler := h.httpHandler
305-
if httpHandler != nil {
306-
httpHandler.server.Stop()
307-
h.httpHandler = nil
308-
}
324+
go func() {
325+
defer close(done)
309326

310-
wsHandler := h.wsHandler
311-
if wsHandler != nil {
312-
wsHandler.server.Stop()
313-
h.wsHandler = nil
314-
}
327+
if h.listener == nil {
328+
return // not running
329+
}
315330

316-
ctx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
317-
defer cancel()
318-
err := h.server.Shutdown(ctx)
319-
if err != nil && err == ctx.Err() {
320-
h.logger.Warn().Msg("HTTP server graceful shutdown timed out")
321-
h.server.Close()
322-
}
331+
// Shut down the server.
332+
httpHandler := h.httpHandler
333+
if httpHandler != nil {
334+
httpHandler.server.Stop()
335+
h.httpHandler = nil
336+
}
337+
338+
wsHandler := h.wsHandler
339+
if wsHandler != nil {
340+
wsHandler.server.Stop()
341+
h.wsHandler = nil
342+
}
343+
344+
ctx, cancel := context.WithTimeout(context.Background(), shutdownTimeout)
345+
defer cancel()
346+
err := h.server.Shutdown(ctx)
347+
if err != nil && err == ctx.Err() {
348+
h.logger.Warn().Msg("HTTP server graceful shutdown timed out")
349+
h.server.Close()
350+
}
323351

324-
h.listener.Close()
325-
h.logger.Info().Msgf(
326-
"HTTP server stopped, endpoint: %s", h.listener.Addr(),
327-
)
352+
h.listener.Close()
353+
h.logger.Info().Msgf(
354+
"HTTP server stopped, endpoint: %s", h.listener.Addr(),
355+
)
356+
357+
// Clear out everything to allow re-configuring it later.
358+
h.host, h.port, h.endpoint = "", 0, ""
359+
h.server, h.listener = nil, nil
360+
361+
}()
328362

329-
// Clear out everything to allow re-configuring it later.
330-
h.host, h.port, h.endpoint = "", 0, ""
331-
h.server, h.listener = nil, nil
363+
return done
332364
}
333365

334366
// CheckTimeouts ensures that timeout values are meaningful

‎api/stream.go

+3-11
Original file line numberDiff line numberDiff line change
@@ -178,7 +178,7 @@ func newSubscription[T any](
178178

179179
rpcSub := notifier.CreateSubscription()
180180

181-
subs := models.NewSubscription(logger, callback(notifier, rpcSub))
181+
subs := models.NewSubscription(callback(notifier, rpcSub))
182182

183183
l := logger.With().
184184
Str("gateway-subscription-id", fmt.Sprintf("%p", subs)).
@@ -190,16 +190,8 @@ func newSubscription[T any](
190190
go func() {
191191
defer publisher.Unsubscribe(subs)
192192

193-
for {
194-
select {
195-
case err := <-subs.Error():
196-
l.Debug().Err(err).Msg("subscription returned error")
197-
return
198-
case err := <-rpcSub.Err():
199-
l.Debug().Err(err).Msg("client unsubscribed")
200-
return
201-
}
202-
}
193+
err := <-rpcSub.Err()
194+
l.Debug().Err(err).Msg("client unsubscribed")
203195
}()
204196

205197
l.Info().Msg("new heads subscription created")

0 commit comments

Comments
 (0)