Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Improved build. .env and config.yaml exposed for this example's purpo…
…ses only
- Loading branch information
1 parent
1ccfabd
commit c32274c
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,4 @@ | ||
| DB_HOST=db | ||
| DB_USER=postgres | ||
| DB_PASSWORD=postgres | ||
| DB_NAME=postgres |
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 |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| db: | ||
| host: "localhost" | ||
| user: "postgres" | ||
| password: "postgres" | ||
| name: "postgres" | ||
| symbols: | ||
| - "btcusdt" | ||
| - "ethusdt" | ||
| - "ltcusdt" |
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 |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package monitor | ||
|
|
||
| import ( | ||
| "database/sql" | ||
| "fmt" | ||
| "log" | ||
| "time" | ||
|
|
||
| "github.com/arturogonzalezm/RealTimeBinanceMonitor/internal/processor" | ||
| "github.com/arturogonzalezm/RealTimeBinanceMonitor/internal/websocket" | ||
| ) | ||
|
|
||
| // MonitorSymbol starts monitoring for a specific symbol | ||
| func MonitorSymbol(symbol string, db *sql.DB, stop chan struct{}) { | ||
| log.Printf("Starting monitoring for symbol: %s", symbol) | ||
| uri := fmt.Sprintf("wss://stream.binance.com:9443/ws/%s@ticker", symbol) | ||
|
|
||
| client := websocket.NewClient() | ||
| if err := client.Connect(uri); err != nil { | ||
| log.Fatalf("WebSocket connection error for symbol %s: %v", symbol, err) | ||
| } | ||
| defer func() { | ||
| if err := client.Close(); err != nil { | ||
| log.Printf("Error closing WebSocket client for symbol %s: %v", symbol, err) | ||
| } | ||
| }() | ||
|
|
||
| // Create PGWriter | ||
| pgWriter, err := processor.NewPGWriter(db) | ||
| if err != nil { | ||
| log.Fatalf("Error creating PostgreSQL writer for symbol %s: %v", symbol, err) | ||
| } | ||
| defer func() { | ||
| if err := pgWriter.Close(); err != nil { | ||
| log.Printf("Error closing PostgreSQL writer for symbol %s: %v", symbol, err) | ||
| } | ||
| }() | ||
|
|
||
| client.AddProcessor(pgWriter) | ||
|
|
||
| go client.Listen(stop) | ||
|
|
||
| log.Printf("WebSocket connection opened for %s", symbol) | ||
|
|
||
| // Periodic summary | ||
| ticker := time.NewTicker(5 * time.Second) | ||
| defer ticker.Stop() | ||
|
|
||
| for { | ||
| select { | ||
| case <-stop: | ||
| log.Printf("Stopping monitoring for symbol: %s", symbol) | ||
| return | ||
| case <-ticker.C: | ||
| // Print a summary every 5 seconds | ||
| log.Printf("Symbol: %s - Last 5 seconds: Processed %d messages", symbol, pgWriter.GetProcessedCount()) | ||
| log.Printf("Symbol: %s - Current buffer size: %d", symbol, pgWriter.GetBufferSize()) | ||
| } | ||
| } | ||
| } |
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,47 @@ | ||
| package monitor | ||
|
|
||
| import ( | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/DATA-DOG/go-sqlmock" | ||
| "github.com/arturogonzalezm/RealTimeBinanceMonitor/internal/processor" | ||
| ) | ||
|
|
||
| func TestMonitorSymbol(t *testing.T) { | ||
| ctrl := gomock.NewController(t) | ||
| defer ctrl.Finish() | ||
|
|
||
| // Mock dependencies | ||
| db, mockDB, err := sqlmock.New() | ||
| if err != nil { | ||
| t.Fatalf("an error '%s' was not expected when opening a stub database connection", err) | ||
| } | ||
| defer db.Close() | ||
|
|
||
| mockWebSocketClient := websocket_mocks.NewMockWebSocketClient(ctrl) | ||
| mockPGWriter := processor_mocks.NewMockDataProcessor(ctrl) | ||
|
|
||
| // Expectations | ||
| mockWebSocketClient.EXPECT().Connect(gomock.Any()).Return(nil) | ||
| mockWebSocketClient.EXPECT().Close().Return(nil) | ||
| mockWebSocketClient.EXPECT().AddProcessor(mockPGWriter) | ||
| mockWebSocketClient.EXPECT().Listen(gomock.Any()).Do(func(stop chan struct{}) { | ||
| time.Sleep(1 * time.Second) | ||
| close(stop) | ||
| }) | ||
|
|
||
| mockPGWriter.EXPECT().Close().Return(nil) | ||
| mockPGWriter.EXPECT().GetProcessedCount().AnyTimes().Return(0) | ||
| mockPGWriter.EXPECT().GetBufferSize().AnyTimes().Return(0) | ||
|
|
||
| stop := make(chan struct{}) | ||
| go MonitorSymbol("btcusdt", db, stop, mockWebSocketClient, mockPGWriter) | ||
|
|
||
| select { | ||
| case <-stop: | ||
| // Test passed | ||
| case <-time.After(2 * time.Second): | ||
| t.Fatalf("Test timed out") | ||
| } | ||
| } | ||
