rss-tools: app/app.go (626175b)

1

package app

2

3

import (

4

	"context"

5

	"fmt"

6

	"log/slog"

7

	"net/http"

8

	"strings"

9

	"sync"

10

	"time"

11

12

	"go.etcd.io/bbolt"

13

14

15

type App struct {

16

	mux           *http.ServeMux

17

	workers       []func(context.Context) error

18

	shutdowns     []func(context.Context)

19

	wg            *sync.WaitGroup

20

	db            *bbolt.DB

21

	scraperClient *http.Client

22

23

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

24

	Config *Config

25

	Client *http.Client

26

	Logger *slog.Logger

27

28

29

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

30

	return &App{

31

		mux:           http.NewServeMux(),

32

		workers:       nil,

33

		wg:            &sync.WaitGroup{},

34

		db:            db,

35

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

36

37

		Logger: slog.Default(),

38

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

39

		Config: cfg,

40

41

42

43

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

44

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

45

	a.mux.HandleFunc(pattern, handler)

46

47

48

// AddWorker adds background worker

49

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

50

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

51

52

53

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

54

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

55

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

56

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

57

58

59

const (

60

	defaultScraperUserAgent = "rss-tools/1.0)"

61

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

62

63

64

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

65

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

66

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

67

	if err != nil {

68

		return nil, err

69

70

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

71

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

72

	return a.scraperClient.Do(req)

73

74

75

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

76

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

77

	// workers

78

	for _, worker := range a.workers {

79

		a.wg.Add(1)

80

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

81

			defer a.wg.Done()

82

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

83

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

84

85

		}(worker)

86

87

88

	// http server

89

	handler := a.recoverMiddleware(a.mux)

90

	handler = a.loggingMiddleware(handler)

91

	if strings.TrimSpace(a.Config.AuthToken) != "" {

92

		handler = a.authMiddleware(handler)

93

94

95

	httpSrv := &http.Server{

96

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

97

		Handler: handler,

98

99

100

	go func() {

101

		go func() {

102

			<-ctx.Done()

103

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

104

			defer cancel()

105

			httpSrv.Shutdown(shutdownCtx)

106

}()

107

}()

108

109

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

110

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

111

		return err

112

113

114

	a.wg.Wait()

115

116

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

117

	defer cancel()

118

	for _, fn := range a.shutdowns {

119

		fn(shutdownCtx)

120

121

122

	return nil

123

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