()
| 17 | var conf = config.New() |
| 18 | |
| 19 | func main() { |
| 20 | fmt.Println("[Database service]") |
| 21 | |
| 22 | // Postgres connection |
| 23 | connPG, err := sql.Open("postgres", conf.PostgresURL+"?sslmode=disable") |
| 24 | if err != nil { |
| 25 | log.Fatalf("postgres connection: %s", err) |
| 26 | } |
| 27 | defer connPG.Close() |
| 28 | |
| 29 | // The table is not created when deployed on AWS RDS. |
| 30 | _, err = connPG.Exec("create table if not exists messages (id serial primary key, message text not null, created timestamp not null)") |
| 31 | if err != nil { |
| 32 | log.Fatalf("create table: %s", err) |
| 33 | } |
| 34 | |
| 35 | // RabbitMQ connection |
| 36 | connMQ, err := rabbit.GetConn(conf.RabbitURL) |
| 37 | if err != nil { |
| 38 | log.Fatalf("rabbit connection: %s", err) |
| 39 | } |
| 40 | defer connMQ.Close() |
| 41 | |
| 42 | err = connMQ.DeclareTopicExchange(conf.Exchange) |
| 43 | if err != nil { |
| 44 | log.Fatalf("declare exchange: %s", err) |
| 45 | } |
| 46 | |
| 47 | // Start a Rabbit consumer with a message processing handler. |
| 48 | connMQ.StartConsumer(conf.Exchange, conf.QueueDB, conf.KeyDB, func(d amqp.Delivery) bool { |
| 49 | return insertToDB(d, connPG) |
| 50 | }) |
| 51 | |
| 52 | select {} |
| 53 | } |
| 54 | |
| 55 | // insertToDB inserts a Rabbit message into a Postgres database. |
| 56 | func insertToDB(d amqp.Delivery, c *sql.DB) bool { |
nothing calls this directly
no test coverage detected