rss-tools: app/app.go (20e54e61e32f5ae335cc7c7015d3969c20462437)

1

package app

2

3

import (

4

	"context"

5

	"fmt"

6

	"log/slog"

7

	"net/http"

8

	"sync"

9

	"time"

10

11

	"go.etcd.io/bbolt"

12

13

14

type App struct {

15

	mux           *http.ServeMux

16

	workers       []func(context.Context) error

17

	shutdowns     []func(context.Context)

18

	wg            *sync.WaitGroup

19

	db            *bbolt.DB

20

	scraperClient *http.Client

21

22

	// TODO: cacher, each scrapper should be able to get it's own cacher

23

	Config *Config

24

	Client *http.Client

25

	Logger *slog.Logger

26

27

28

func New(cfg *Config, db *bbolt.DB) *App {

29

	return &App{

30

		mux:           http.NewServeMux(),

31

		workers:       nil,

32

		wg:            &sync.WaitGroup{},

33

		db:            db,

34

		scraperClient: &http.Client{Timeout: 10 * time.Second},

35

36

		Logger: slog.Default(),

37

		Client: &http.Client{Timeout: 31 * time.Second},

38

		Config: cfg,

39

40

41

42

// Route registers a global route. pattern syntax is the same as in [http.ServeMux].HandlerFunc

43

func (a App) Route(pattern string, handler http.HandlerFunc) {

44

	a.mux.HandleFunc(pattern, handler)

45

46

47

// AddWorker adds background worker

48

func (a *App) AddWorker(worker func(ctx context.Context) error) {

49

	a.workers = append(a.workers, worker)

50

51

52

// AddShutdown registers a shutdown hook that will be called when the app stops.

53

// Shutdown hooks are called in reverse order of registration.

54

func (a *App) AddShutdown(fn func(ctx context.Context)) {

55

	a.shutdowns = append(a.shutdowns, fn)

56

57

58

const (

59

	defaultScraperUserAgent = "rss-tools/1.0)"

60

	defaultScraperAccept    = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"

61

62

63

// Get is intended for scraping sources; API SDK calls should use [App.Client] directly.

64

func (a *App) Get(ctx context.Context, url string) (*http.Response, error) {

65

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)

66

	if err != nil {

67

		return nil, err

68

69

	req.Header.Set("User-Agent", defaultScraperUserAgent)

70

	req.Header.Set("Accept", defaultScraperAccept)

71

	return a.scraperClient.Do(req)

72

73

74

// Start starts an app and with all it's registered sources

75

func (a *App) Start(ctx context.Context) error {

76

	// workers

77

	for _, worker := range a.workers {

78

		a.wg.Add(1)

79

		go func(w func(context.Context) error) {

80

			defer a.wg.Done()

81

			if err := w(ctx); err != nil {

82

				slog.ErrorContext(ctx, "worker exited with an error", "err", err)

83

84

		}(worker)

85

86

87

	// http server

88

	handler := a.recoverMiddleware(a.mux)

89

	handler = a.authMiddleware(handler)

90

	handler = a.loggingMiddleware(handler)

91

	httpSrv := &http.Server{

92

		Addr:    fmt.Sprintf(":%d", a.Config.Port), // fixme

93

		Handler: handler,

94

95

96

	go func() {

97

		go func() {

98

			<-ctx.Done()

99

			shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

100

			defer cancel()

101

			httpSrv.Shutdown(shutdownCtx)

102

}()

103

}()

104

105

	slog.Info("starting http server", "port", a.Config.Port)

106

	if err := httpSrv.ListenAndServe(); err != http.ErrServerClosed {

107

		return err

108

109

110

	a.wg.Wait()

111

112

	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

113

	defer cancel()

114

	for _, fn := range a.shutdowns {

115

		fn(shutdownCtx)

116

117

118

	return nil

119

1	package app
2
3	import (
4	"context"
5	"fmt"
6	"log/slog"
7	"net/http"
8	"sync"
9	"time"
10
11	"go.etcd.io/bbolt"
12	)
13
14	type App struct {
15	mux *http.ServeMux
16	workers []func(context.Context) error
17	shutdowns []func(context.Context)
18	wg *sync.WaitGroup
19	db *bbolt.DB
20	scraperClient *http.Client
21
22	// TODO: cacher, each scrapper should be able to get it's own cacher
23	Config *Config
24	Client *http.Client
25	Logger *slog.Logger
26	}
27
28	func New(cfg Config, db bbolt.DB) *App {
29	return &App{
30	mux: http.NewServeMux(),
31	workers: nil,
32	wg: &sync.WaitGroup{},
33	db: db,
34	scraperClient: &http.Client{Timeout: 10 * time.Second},
35
36	Logger: slog.Default(),
37	Client: &http.Client{Timeout: 31 * time.Second},
38	Config: cfg,
39	}
40	}
41
42	// Route registers a global route. pattern syntax is the same as in [http.ServeMux].HandlerFunc
43	func (a App) Route(pattern string, handler http.HandlerFunc) {
44	a.mux.HandleFunc(pattern, handler)
45	}
46
47	// AddWorker adds background worker
48	func (a *App) AddWorker(worker func(ctx context.Context) error) {
49	a.workers = append(a.workers, worker)
50	}
51
52	// AddShutdown registers a shutdown hook that will be called when the app stops.
53	// Shutdown hooks are called in reverse order of registration.
54	func (a *App) AddShutdown(fn func(ctx context.Context)) {
55	a.shutdowns = append(a.shutdowns, fn)
56	}
57
58	const (
59	defaultScraperUserAgent = "rss-tools/1.0)"
60	defaultScraperAccept = "text/html,application/xhtml+xml,application/xml;q=0.9,/;q=0.8"
61	)
62
63	// Get is intended for scraping sources; API SDK calls should use [App.Client] directly.
64	func (a App) Get(ctx context.Context, url string) (http.Response, error) {
65	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
66	if err != nil {
67	return nil, err
68	}
69	req.Header.Set("User-Agent", defaultScraperUserAgent)
70	req.Header.Set("Accept", defaultScraperAccept)
71	return a.scraperClient.Do(req)
72	}
73
74	// Start starts an app and with all it's registered sources
75	func (a *App) Start(ctx context.Context) error {
76	// workers
77	for _, worker := range a.workers {
78	a.wg.Add(1)
79	go func(w func(context.Context) error) {
80	defer a.wg.Done()
81	if err := w(ctx); err != nil {
82	slog.ErrorContext(ctx, "worker exited with an error", "err", err)
83	}
84	}(worker)
85	}
86
87	// http server
88	handler := a.recoverMiddleware(a.mux)
89	handler = a.authMiddleware(handler)
90	handler = a.loggingMiddleware(handler)
91	httpSrv := &http.Server{
92	Addr: fmt.Sprintf(":%d", a.Config.Port), // fixme
93	Handler: handler,
94	}
95
96	go func() {
97	go func() {
98	<-ctx.Done()
99	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
100	defer cancel()
101	httpSrv.Shutdown(shutdownCtx)
102	}()
103	}()
104
105	slog.Info("starting http server", "port", a.Config.Port)
106	if err := httpSrv.ListenAndServe(); err != http.ErrServerClosed {
107	return err
108	}
109
110	a.wg.Wait()
111
112	shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
113	defer cancel()
114	for _, fn := range a.shutdowns {
115	fn(shutdownCtx)
116	}
117
118	return nil
119	}