generated from rog-golang-buddies/golang-template-repository
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #17 from rog-golang-buddies/queue_setup
Queue setup
- Loading branch information
Showing
25 changed files
with
844 additions
and
14 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,8 +1,10 @@ | ||
| package main | ||
|
|
||
| import "fmt" | ||
| import ( | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal" | ||
| "os" | ||
| ) | ||
|
|
||
| func main() { | ||
| // Feel free to delete this file. | ||
| fmt.Println("Hello Gophers") | ||
| os.Exit(internal.Start()) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| version: '3.9' | ||
|
|
||
| services: | ||
| rabbit: | ||
| image: rabbitmq:3-management #you may open management UI via http://localhost:15672/#/ login&password == guest | ||
| container_name: rabbit | ||
| ports: | ||
| - "5672:5672" | ||
| - "15672:15672" | ||
|
|
||
| data-scraping-service: | ||
| container_name: dss | ||
| build: | ||
| context: ../. | ||
| dockerfile: Dockerfile | ||
| restart: unless-stopped | ||
| depends_on: | ||
| - rabbit |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,16 @@ | ||
| module github.com/rog-golang-buddies/api-hub_data-scraping-service | ||
|
|
||
| go 1.18 | ||
|
|
||
| require ( | ||
| github.com/golang/mock v1.6.0 | ||
| github.com/rabbitmq/amqp091-go v1.4.0 | ||
| github.com/stretchr/testify v1.7.0 | ||
| github.com/wagslane/go-rabbitmq v0.10.0 | ||
| ) | ||
|
|
||
| require ( | ||
| github.com/davecgh/go-spew v1.1.0 // indirect | ||
| github.com/pmezard/go-difflib v1.0.0 // indirect | ||
| gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c // indirect | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| package internal | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/config" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/queue" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/queue/handler" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/queue/publisher" | ||
| "log" | ||
| ) | ||
|
|
||
| func Start() int { | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| defer cancel() | ||
|
|
||
| conf := config.ReadConfig() //read configuration from file & env | ||
| //initialize publisher connection to the queue | ||
| //this library assumes using one publisher and one consumer per application | ||
| //https://github.com/wagslane/go-rabbitmq/issues/79 | ||
| pub, err := publisher.NewPublisher(conf.QueueConfig) //TODO pass logger here and add it to publisher options | ||
| if err != nil { | ||
| log.Println("error while starting publisher: ", err) | ||
| return 1 | ||
| } | ||
| defer publisher.ClosePublisher(pub) | ||
| //initialize consumer connection to the queue | ||
| consumer, err := queue.NewConsumer(conf.QueueConfig) //TODO pass logger here and add it to consumer options | ||
| if err != nil { | ||
| log.Println("error while connecting to the queue: ", err) | ||
| return 1 | ||
| } | ||
| defer queue.CloseConsumer(consumer) | ||
|
|
||
| handl := handler.NewApiSpecDocHandler(pub, conf.QueueConfig) | ||
| listener := queue.NewListener() | ||
| err = listener.Start(consumer, &conf.QueueConfig, handl) | ||
| if err != nil { | ||
| log.Println("error while listening queue ", err) | ||
| return 1 | ||
| } | ||
|
|
||
| <-ctx.Done() | ||
|
|
||
| log.Println("application stopped gracefully (not)") | ||
| return 0 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,20 @@ | ||
| package config | ||
|
|
||
| type ApplicationConfig struct { | ||
| QueueConfig QueueConfig | ||
| } | ||
|
|
||
| func ReadConfig() ApplicationConfig { | ||
| //Stub this method before the configuration task is not resolved | ||
| //https://github.com/rog-golang-buddies/api-hub_data-scraping-service/issues/10 | ||
| //TODO implement with the method to read configuration from file and env | ||
| return ApplicationConfig{ | ||
| QueueConfig: QueueConfig{ | ||
| UrlRequestQueue: "data-scraping-asd", | ||
| ScrapingResultQueue: "storage-update-asd", | ||
| NotificationQueue: "gateway-scrape_notifications", | ||
| Url: "amqp://guest:guest@rabbit:5672/", | ||
| Concurrency: 10, | ||
| }, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| package config | ||
|
|
||
| //QueueConfig queue configuration | ||
| type QueueConfig struct { | ||
| UrlRequestQueue string //UrlRequestQueue name to listen to the new events | ||
| ScrapingResultQueue string //Queue name to send processed ApiSpecDoc | ||
| NotificationQueue string //Queue name to notify a user about error or success (if required) | ||
| Url string //RabbitMQ url | ||
| Concurrency int //Number of parallel handlers | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| package dto | ||
|
|
||
| import "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/dto/apiSpecDoc" | ||
|
|
||
| type ScrapingResult struct { | ||
| IsNotifyUser bool | ||
|
|
||
| ApiSpecDoc apiSpecDoc.ApiSpecDoc | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| package dto | ||
|
|
||
| //UrlRequest represents listening request model | ||
| type UrlRequest struct { | ||
| //File url to scrape data | ||
| FileUrl string | ||
|
|
||
| //A flag is a notification required related to an error notification in case of an error | ||
| //Notification is required when this is the request from the user and doesn't require it | ||
| //if it is the request from the storage and update service. | ||
| IsNotifyUser bool | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,32 @@ | ||
| package dto | ||
|
|
||
| import "fmt" | ||
|
|
||
| //UserNotification represents basic DTO notification to the user if requested | ||
| //Initially, it supposed to be simple - if err != nil => error happens, else all is ok | ||
| type UserNotification struct { | ||
| Error *ProcessingError | ||
| } | ||
|
|
||
| func NewUserNotification(procErr *ProcessingError) UserNotification { | ||
| return UserNotification{Error: procErr} | ||
| } | ||
|
|
||
| //ProcessingError represents basic DTO to provide information about the error | ||
| //when the processing request contains a notification request | ||
| type ProcessingError struct { | ||
| Cause error | ||
|
|
||
| Message string | ||
| } | ||
|
|
||
| func (pe *ProcessingError) Error() string { | ||
| return fmt.Sprintf("%s: %v", pe.Message, pe.Cause) | ||
| } | ||
|
|
||
| func NewProcessingError(message string, err error) ProcessingError { | ||
| return ProcessingError{ | ||
| Cause: err, | ||
| Message: message, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,40 @@ | ||
| package queue | ||
|
|
||
| import ( | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/config" | ||
| "github.com/wagslane/go-rabbitmq" | ||
| "io" | ||
| "log" | ||
| ) | ||
|
|
||
| //Consumer is just an interface for the library consumer which doesn't have one. | ||
| //go:generate mockgen -source=consumer.go -destination=./mocks/consumer.go | ||
| type Consumer interface { | ||
| io.Closer | ||
| StartConsuming( | ||
| handler rabbitmq.Handler, | ||
| queue string, | ||
| routingKeys []string, | ||
| optionFuncs ...func(*rabbitmq.ConsumeOptions), | ||
| ) error | ||
| } | ||
|
|
||
| func NewConsumer(conf config.QueueConfig) (Consumer, error) { | ||
| consumer, err := rabbitmq.NewConsumer( | ||
| conf.Url, | ||
| rabbitmq.Config{}, | ||
| rabbitmq.WithConsumerOptionsLogging, | ||
| ) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return &consumer, nil | ||
| } | ||
|
|
||
| func CloseConsumer(consumer Consumer) { | ||
| log.Println("closing consumer") | ||
| err := consumer.Close() | ||
| if err != nil { | ||
| log.Println("error while closing consumer: ", err) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| package queue_test | ||
|
|
||
| import ( | ||
| "github.com/golang/mock/gomock" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/queue" | ||
| mock_queue "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/queue/mocks" | ||
| "testing" | ||
| ) | ||
|
|
||
| func TestClosePublisher(t *testing.T) { | ||
| ctrl := gomock.NewController(t) | ||
| consumer := mock_queue.NewMockConsumer(ctrl) | ||
| consumer.EXPECT().Close().Return(nil) | ||
| queue.CloseConsumer(consumer) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| package handler | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/config" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/dto" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/dto/apiSpecDoc" | ||
| "github.com/rog-golang-buddies/api-hub_data-scraping-service/internal/queue/publisher" | ||
| "github.com/wagslane/go-rabbitmq" | ||
| "log" | ||
| ) | ||
|
|
||
| type ApiSpecDocHandler struct { | ||
| publisher publisher.Publisher | ||
| config config.QueueConfig | ||
| } | ||
|
|
||
| func (asdh *ApiSpecDocHandler) Handle(delivery rabbitmq.Delivery) rabbitmq.Action { | ||
| log.Printf("consumed: %v", string(delivery.Body)) | ||
| //call process here | ||
| var req dto.UrlRequest | ||
| err := json.Unmarshal(delivery.Body, &req) | ||
| if err != nil { | ||
| log.Printf("error unmarshalling message: '%v', err: %s\n", string(delivery.Body), err) | ||
| return rabbitmq.NackDiscard | ||
| } | ||
| //here processing of the request happens... | ||
| asd := apiSpecDoc.ApiSpecDoc{} //TODO replace this stub with process call | ||
|
|
||
| //publish to the required queue success or error | ||
| result := dto.ScrapingResult{IsNotifyUser: req.IsNotifyUser, ApiSpecDoc: asd} | ||
| err = asdh.publish(&delivery, result, asdh.config.ScrapingResultQueue) | ||
| if err != nil { | ||
| log.Println("error while publishing: ", err) | ||
| //Here is some error while publishing happened - probably something wrong with the queue | ||
| return rabbitmq.NackDiscard | ||
| } | ||
| if req.IsNotifyUser { | ||
| err = asdh.publish(&delivery, dto.NewUserNotification(nil), asdh.config.NotificationQueue) | ||
| if err != nil { | ||
| log.Println("error while notifying user") | ||
| //don't discard this message because it was published to the storage service successfully | ||
| } | ||
| } | ||
| log.Println("Url scraped successfully") | ||
| return rabbitmq.Ack | ||
| } | ||
|
|
||
| func (asdh *ApiSpecDocHandler) publish(delivery *rabbitmq.Delivery, message any, queue string) error { | ||
| content, err := json.Marshal(message) | ||
| if err != nil { | ||
| log.Println("error while marshalling: ", err) | ||
| return err | ||
| } | ||
| return asdh.publisher.Publish(content, | ||
| []string{queue}, | ||
| rabbitmq.WithPublishOptionsCorrelationID(delivery.CorrelationId), | ||
| rabbitmq.WithPublishOptionsContentType("application/json"), | ||
| rabbitmq.WithPublishOptionsPersistentDelivery, | ||
| ) | ||
| } | ||
|
|
||
| func NewApiSpecDocHandler(publisher publisher.Publisher, config config.QueueConfig) Handler { | ||
| return &ApiSpecDocHandler{ | ||
| publisher: publisher, | ||
| config: config, | ||
| } | ||
| } |
Oops, something went wrong.