-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.go
executable file
·188 lines (153 loc) · 5.81 KB
/
main.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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
package main
import (
"context"
"net"
"net/http"
"os"
"github.com/hibiken/asynq"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
"github.com/golang-migrate/migrate/v4"
_ "github.com/golang-migrate/migrate/v4/database/postgres"
_ "github.com/golang-migrate/migrate/v4/source/file"
"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
"github.com/hoangtk0100/simple-bank/api"
db "github.com/hoangtk0100/simple-bank/db/sqlc"
_ "github.com/hoangtk0100/simple-bank/docs/statik"
"github.com/hoangtk0100/simple-bank/gapi"
"github.com/hoangtk0100/simple-bank/mail"
"github.com/hoangtk0100/simple-bank/pb"
"github.com/hoangtk0100/simple-bank/util"
"github.com/hoangtk0100/simple-bank/worker"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/rakyll/statik/fs"
"google.golang.org/grpc"
"google.golang.org/grpc/reflection"
"google.golang.org/protobuf/encoding/protojson"
)
func main() {
config, err := util.LoadConfig(".")
if err != nil {
log.Fatal().Err(err).Msg("Cannot load config")
}
if config.Environment == "development" {
log.Logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr})
}
connPool, err := pgxpool.New(context.Background(), config.DBSource)
if err != nil {
log.Fatal().Err(err).Msg("Cannot connect to DB")
}
runDBMigration(config.MigrationURL, config.DBSource)
store := db.NewStore(connPool)
redisOpt := asynq.RedisClientOpt{
Addr: config.RedisAddress,
Password: config.RedisPassword,
}
taskDistributor := worker.NewRedisTaskDistributor(redisOpt)
// runTaskProcessor in a separate go routine
// Because when the processor starts, the Asynq server will block and keep polling Redis for new tasks
go runTaskProcessor(config, redisOpt, store)
// Can not call both runGrpcServer, runGinServer for serving both GRPC and HTTP requests in the same go routine
// so run 1 of them in the separate go routine, not blocking each other from starting
go runGatewayServer(config, store, taskDistributor)
runGrpcServer(config, store, taskDistributor)
}
func runDBMigration(migrationURL string, dbSource string) {
migration, err := migrate.New(migrationURL, dbSource)
if err != nil {
log.Fatal().Err(err).Msg("can not create new migrate instance")
}
if err = migration.Up(); err != nil && err != migrate.ErrNoChange {
log.Fatal().Err(err).Msg("failed to run migrate up")
}
log.Info().Msg("db migrated successfully")
}
func runTaskProcessor(config util.Config, redisOpt asynq.RedisClientOpt, store db.Store) {
mailer := mail.NewGmailSender(config.EmailSenderName, config.EmailSenderAddress, config.EmailSenderPassword)
taskProcessor := worker.NewRedisTaskProcessor(redisOpt, store, mailer)
log.Info().Msg("start task processor")
err := taskProcessor.Start()
if err != nil {
log.Fatal().Err(err).Msg("failed to start task processor")
}
}
// runGatewayServer: Set up HTTP gateway with in-process translation method
func runGatewayServer(config util.Config, store db.Store, taskDistributor worker.TaskDistributor) {
server, err := gapi.NewServer(config, store, taskDistributor)
if err != nil {
log.Fatal().Err(err).Msg("can not create server")
}
// To map response field names the same style in proto files (default: camelCase)
jsonOption := runtime.WithMarshalerOption(runtime.MIMEWildcard, &runtime.JSONPb{
MarshalOptions: protojson.MarshalOptions{
UseProtoNames: true,
},
UnmarshalOptions: protojson.UnmarshalOptions{
DiscardUnknown: true,
},
})
grpcMux := runtime.NewServeMux(jsonOption)
ctx, cancel := context.WithCancel(context.Background())
// only executed before exiting this runGatewayServer function
// cancelling a context is a way to prevent the system from doing unnecessary works
defer cancel()
err = pb.RegisterSimpleBankHandlerServer(ctx, grpcMux, server)
if err != nil {
log.Fatal().Err(err).Msg("can not register handler server")
}
// Receive HTTP requests from the clients
mux := http.NewServeMux()
// To convert HTTP request to gRPC format, reroute them to the gRPC mux
mux.Handle("/", grpcMux)
// Serve swagger requests from files in directory
// fs := http.FileServer(http.Dir("./docs/swagger"))
// mux.Handle("/swagger/", http.StripPrefix("/swagger/", fs))
// Serve swagger from statik binary file
// New() use default namespace
// NewWithNamespace() for using custom namespace
statikFS, err := fs.New()
if err != nil {
log.Fatal().Err(err).Msg("can not create statik fs")
}
swaggerHandler := http.StripPrefix("/swagger/", http.FileServer(statikFS))
mux.Handle("/swagger/", swaggerHandler)
listener, err := net.Listen("tcp", config.HTTPServerAddress)
if err != nil {
log.Fatal().Err(err).Msg("can not create listener")
}
log.Info().Msgf("start HTTP gateway server at %s", listener.Addr().String())
handler := gapi.HttpLogger(mux)
err = http.Serve(listener, handler)
if err != nil {
log.Fatal().Err(err).Msg("can not start HTTP gateway server")
}
}
func runGrpcServer(config util.Config, store db.Store, taskDistributor worker.TaskDistributor) {
server, err := gapi.NewServer(config, store, taskDistributor)
if err != nil {
log.Fatal().Err(err).Msg("can not create server")
}
grpcLogger := grpc.UnaryInterceptor(gapi.GrpcLogger)
grpcServer := grpc.NewServer(grpcLogger)
pb.RegisterSimpleBankServer(grpcServer, server)
reflection.Register(grpcServer)
listener, err := net.Listen("tcp", config.GRPCServerAddress)
if err != nil {
log.Fatal().Err(err).Msg("can not create listener:")
}
log.Info().Msgf("start gRPC server at %s", listener.Addr().String())
err = grpcServer.Serve(listener)
if err != nil {
log.Fatal().Err(err).Msg("can not start gRPC server")
}
}
func runGinServer(config util.Config, store db.Store) {
server, err := api.NewServer(config, store)
if err != nil {
log.Fatal().Err(err).Msg("can not create server")
}
err = server.Start(config.HTTPServerAddress)
if err != nil {
log.Fatal().Err(err).Msg("Cannot start server")
}
}