Compare commits
2 Commits
dev
...
cashe-more
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
497b5a6e86 | ||
|
|
8b56bf8370 |
@@ -8,13 +8,11 @@
|
|||||||
"sslmode": "disable"
|
"sslmode": "disable"
|
||||||
},
|
},
|
||||||
"valkey": {
|
"valkey": {
|
||||||
"host": "127.0.0.1",
|
"host": "127.0.0.1",
|
||||||
"port": "6379",
|
"port": "6379",
|
||||||
"max_conns": 50,
|
"max_conns": 50,
|
||||||
"timeout_ms": 5000,
|
"timeout_ms": 5000,
|
||||||
"password": "the_valkey_password"
|
"password": "the_valkey_password"
|
||||||
},
|
},
|
||||||
"temp": "value",
|
"temp": "value"
|
||||||
"dataset_id": "",
|
|
||||||
"excluded_dataset_ids": ""
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,8 +7,6 @@ import (
|
|||||||
"strconv"
|
"strconv"
|
||||||
)
|
)
|
||||||
|
|
||||||
const configFilePath = "config/conf.json"
|
|
||||||
|
|
||||||
type Config struct {
|
type Config struct {
|
||||||
Database struct {
|
Database struct {
|
||||||
Host string `json:"host"`
|
Host string `json:"host"`
|
||||||
@@ -23,16 +21,14 @@ type Config struct {
|
|||||||
Port string `json:"port"`
|
Port string `json:"port"`
|
||||||
MaxConns int `json:"max_conns"`
|
MaxConns int `json:"max_conns"`
|
||||||
TimeoutMs int `json:"timeout_ms"`
|
TimeoutMs int `json:"timeout_ms"`
|
||||||
Password string `json:"password"`
|
Password string `json:"password"` // Add this line
|
||||||
} `json:"valkey"`
|
} `json:"valkey"`
|
||||||
Temp string `json:"temp"`
|
Temp string `json:"temp"`
|
||||||
DatasetId string `json:"dataset_id"`
|
|
||||||
ExcludedDatasetIds string `json:"excluded_dataset_ids"`
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func LoadConfig() (Config, error) {
|
func LoadConfig(file string) (Config, error) {
|
||||||
var config Config
|
var config Config
|
||||||
configFile, err := os.Open(configFilePath)
|
configFile, err := os.Open(file)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return config, fmt.Errorf("failed to open config file: %w", err)
|
return config, fmt.Errorf("failed to open config file: %w", err)
|
||||||
}
|
}
|
||||||
@@ -84,13 +80,5 @@ func LoadConfig() (Config, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Override datasetId and excludedDatasetIds with environment variables
|
|
||||||
if datasetId := os.Getenv("DATASET_ID"); datasetId != "" {
|
|
||||||
config.DatasetId = datasetId
|
|
||||||
}
|
|
||||||
if excludedDatasetIds := os.Getenv("EXCLUDED_DATASET_IDS"); excludedDatasetIds != "" {
|
|
||||||
config.ExcludedDatasetIds = excludedDatasetIds
|
|
||||||
}
|
|
||||||
|
|
||||||
return config, nil
|
return config, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -11,7 +11,7 @@ import (
|
|||||||
|
|
||||||
func ConnectToPostgreSQL() (*sql.DB, error) {
|
func ConnectToPostgreSQL() (*sql.DB, error) {
|
||||||
fmt.Println("Connecting to PostgreSQL...")
|
fmt.Println("Connecting to PostgreSQL...")
|
||||||
config, err := LoadConfig()
|
config, err := LoadConfig("config/conf.json")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -57,7 +57,7 @@ func DisconnectFromPostgreSQL(db *sql.DB) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func PrintDBConfig() {
|
func PrintDBConfig() {
|
||||||
config, err := LoadConfig()
|
config, err := LoadConfig("config/conf.json")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Println("Error loading config:", err)
|
fmt.Println("Error loading config:", err)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ type ValkeyConfig struct {
|
|||||||
Port string `json:"port"`
|
Port string `json:"port"`
|
||||||
MaxConns int `json:"max_conns"`
|
MaxConns int `json:"max_conns"`
|
||||||
TimeoutMs int `json:"timeout_ms"`
|
TimeoutMs int `json:"timeout_ms"`
|
||||||
Password string `json:"password"`
|
Password string `json:"password"` // Add this line
|
||||||
}
|
}
|
||||||
|
|
||||||
func LoadValkeyConfig(file string) (ValkeyConfig, error) {
|
func LoadValkeyConfig(file string) (ValkeyConfig, error) {
|
||||||
@@ -56,9 +56,9 @@ func LoadValkeyConfig(file string) (ValkeyConfig, error) {
|
|||||||
return config, nil
|
return config, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func ConnectToValkey() (valkey.Client, error) {
|
func ConnectToValkey(configPath string) (valkey.Client, error) {
|
||||||
fmt.Println("Loading configuration...")
|
fmt.Println("Loading configuration...")
|
||||||
config, err := LoadConfig()
|
config, err := LoadConfig(configPath)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("failed to load config: %v", err)
|
return nil, fmt.Errorf("failed to load config: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
18
data/data.go
18
data/data.go
@@ -1,10 +1,9 @@
|
|||||||
package data
|
package data
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"encoding/xml"
|
|
||||||
"log"
|
"log"
|
||||||
|
"encoding/xml"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strings"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
type Data struct {
|
type Data struct {
|
||||||
@@ -127,20 +126,13 @@ type Data struct {
|
|||||||
} `xml:"ServiceDelivery"`
|
} `xml:"ServiceDelivery"`
|
||||||
}
|
}
|
||||||
|
|
||||||
func FetchData(timestamp, datasetId, excludedDatasetIds string) (*Data, error) {
|
func FetchData(timestamp string) (*Data, error) {
|
||||||
client := &http.Client{}
|
client := &http.Client{}
|
||||||
requestorId := "ti1-" + timestamp
|
requestorId := "ti1-" + timestamp
|
||||||
|
|
||||||
baseURL := "https://api.entur.io/realtime/v1/rest/et?useOriginalId=true&maxSize=100000&requestorId=" + requestorId
|
url := "https://api.entur.io/realtime/v1/rest/et?useOriginalId=true&maxSize=100000&requestorId=" + requestorId
|
||||||
|
log.Println("Fetching data from URL:", url)
|
||||||
if datasetId != "" {
|
resp, err := client.Get(url)
|
||||||
baseURL += "&datasetId=" + datasetId
|
|
||||||
} else if excludedDatasetIds != "" {
|
|
||||||
baseURL += "&excludedDatasetIds=" + strings.ReplaceAll(excludedDatasetIds, ",", "&excludedDatasetIds=")
|
|
||||||
}
|
|
||||||
|
|
||||||
log.Println("Fetching data from URL:", baseURL)
|
|
||||||
resp, err := client.Get(baseURL)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -3,37 +3,13 @@ package database
|
|||||||
import (
|
import (
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"fmt"
|
"fmt"
|
||||||
"ti1/config"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func GetDatasetVariable(config config.Config) string {
|
|
||||||
if config.DatasetId != "" {
|
|
||||||
fmt.Println(config.DatasetId)
|
|
||||||
return config.DatasetId
|
|
||||||
} else if config.ExcludedDatasetIds != "" {
|
|
||||||
result := "EX." + config.ExcludedDatasetIds
|
|
||||||
fmt.Println(result)
|
|
||||||
return result
|
|
||||||
}
|
|
||||||
fmt.Println("")
|
|
||||||
return ""
|
|
||||||
}
|
|
||||||
|
|
||||||
func InsertServiceDelivery(db *sql.DB, responseTimestamp string, recordedAtTime string) (int, error) {
|
func InsertServiceDelivery(db *sql.DB, responseTimestamp string, recordedAtTime string) (int, error) {
|
||||||
fmt.Println("Inserting ServiceDelivery...")
|
fmt.Println("Inserting ServiceDelivery...")
|
||||||
var id int
|
var id int
|
||||||
|
|
||||||
// Load configuration
|
err := db.QueryRow("INSERT INTO public.ServiceDelivery (ResponseTimestamp, RecordedAtTime) VALUES ($1, $2) RETURNING ID", responseTimestamp, recordedAtTime).Scan(&id)
|
||||||
config, err := config.LoadConfig()
|
|
||||||
if err != nil {
|
|
||||||
fmt.Println("Error loading config:", err)
|
|
||||||
return 0, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Get dataset variable
|
|
||||||
datasetVariable := GetDatasetVariable(config)
|
|
||||||
|
|
||||||
err = db.QueryRow("INSERT INTO public.ServiceDelivery (ResponseTimestamp, RecordedAtTime, Source) VALUES ($1, $2, $3) RETURNING ID", responseTimestamp, recordedAtTime, datasetVariable).Scan(&id)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
fmt.Println(err)
|
fmt.Println(err)
|
||||||
return 0, err
|
return 0, err
|
||||||
|
|||||||
@@ -68,7 +68,6 @@ func SetupDB() error {
|
|||||||
id INTEGER PRIMARY KEY DEFAULT nextval('public.servicedelivery_id_seq'),
|
id INTEGER PRIMARY KEY DEFAULT nextval('public.servicedelivery_id_seq'),
|
||||||
responsetimestamp TIMESTAMPTZ,
|
responsetimestamp TIMESTAMPTZ,
|
||||||
recordedattime TIMESTAMPTZ,
|
recordedattime TIMESTAMPTZ,
|
||||||
source VARCHAR,
|
|
||||||
data JSON
|
data JSON
|
||||||
);`,
|
);`,
|
||||||
}
|
}
|
||||||
|
|||||||
61
docker-compose.yaml
Normal file
61
docker-compose.yaml
Normal file
@@ -0,0 +1,61 @@
|
|||||||
|
services:
|
||||||
|
db:
|
||||||
|
image: postgres:17.2
|
||||||
|
container_name: postgres-db
|
||||||
|
environment:
|
||||||
|
POSTGRES_USER: postgres
|
||||||
|
POSTGRES_PASSWORD: RootPassword
|
||||||
|
POSTGRES_DB: ti1
|
||||||
|
ports:
|
||||||
|
- "5432:5432"
|
||||||
|
volumes:
|
||||||
|
- /tmp/ti1/postgres_data:/var/lib/postgresql/data:z
|
||||||
|
- /tmp/ti1/init.sql:/docker-entrypoint-initdb.d/init.sql:ro,z
|
||||||
|
networks:
|
||||||
|
- app-network
|
||||||
|
healthcheck:
|
||||||
|
test: ["CMD", "pg_isready", "-U", "postgres", "-d", "ti1", "-h", "db"]
|
||||||
|
interval: 10s
|
||||||
|
retries: 5
|
||||||
|
restart: always
|
||||||
|
|
||||||
|
valkey:
|
||||||
|
image: valkey/valkey:latest
|
||||||
|
container_name: valkey
|
||||||
|
environment:
|
||||||
|
VALKEY_PASSWORD: the_valkey_password
|
||||||
|
ports:
|
||||||
|
- "6379:6379"
|
||||||
|
volumes:
|
||||||
|
- /tmp/ti1/valkey_data:/data:z
|
||||||
|
networks:
|
||||||
|
- app-network
|
||||||
|
restart: always
|
||||||
|
|
||||||
|
ti1-container:
|
||||||
|
build:
|
||||||
|
context: .
|
||||||
|
dockerfile: Dockerfile
|
||||||
|
container_name: ti1-container
|
||||||
|
environment:
|
||||||
|
DB_HOST: db
|
||||||
|
DB_PORT: 5432
|
||||||
|
DB_USER: postgres
|
||||||
|
DB_PASSWORD: RootPassword
|
||||||
|
DB_NAME: ti1
|
||||||
|
DB_SSLMODE: disable
|
||||||
|
VALKEY_HOST: valkey
|
||||||
|
VALKEY_PORT: 6379
|
||||||
|
VALKEY_PASSWORD: the_valkey_password
|
||||||
|
depends_on:
|
||||||
|
db:
|
||||||
|
condition: service_healthy
|
||||||
|
valkey:
|
||||||
|
condition: service_started
|
||||||
|
networks:
|
||||||
|
- app-network
|
||||||
|
restart: always
|
||||||
|
|
||||||
|
networks:
|
||||||
|
app-network:
|
||||||
|
driver: bridge
|
||||||
@@ -22,7 +22,7 @@ func DBData(data *data.Data) {
|
|||||||
defer db.Close()
|
defer db.Close()
|
||||||
|
|
||||||
// Connect to Valkey
|
// Connect to Valkey
|
||||||
valkeyClient, err := config.ConnectToValkey()
|
valkeyClient, err := config.ConnectToValkey("config/conf.json")
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("Failed to connect to Valkey: %v", err)
|
log.Fatalf("Failed to connect to Valkey: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
11
main.go
11
main.go
@@ -2,7 +2,6 @@ package main
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"log"
|
"log"
|
||||||
"ti1/config"
|
|
||||||
"ti1/data"
|
"ti1/data"
|
||||||
"ti1/database"
|
"ti1/database"
|
||||||
"ti1/export"
|
"ti1/export"
|
||||||
@@ -13,14 +12,8 @@ func main() {
|
|||||||
log.Println("ti1 v0.2.1")
|
log.Println("ti1 v0.2.1")
|
||||||
log.Println("Starting...")
|
log.Println("Starting...")
|
||||||
|
|
||||||
// Load configuration
|
|
||||||
cfg, err := config.LoadConfig()
|
|
||||||
if err != nil {
|
|
||||||
log.Fatalf("Failed to load config: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Setup the database
|
// Setup the database
|
||||||
err = database.SetupDB()
|
err := database.SetupDB()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatalf("Database setup failed: %v", err)
|
log.Fatalf("Database setup failed: %v", err)
|
||||||
}
|
}
|
||||||
@@ -32,7 +25,7 @@ func main() {
|
|||||||
for {
|
for {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
|
|
||||||
data, err := data.FetchData(starttimestamp, cfg.DatasetId, cfg.ExcludedDatasetIds)
|
data, err := data.FetchData(starttimestamp)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Fatal(err)
|
log.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -9,13 +9,15 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func SetValkeyValue(ctx context.Context, client valkey.Client, key, value string) error {
|
func SetValkeyValue(ctx context.Context, client valkey.Client, key, value string) error {
|
||||||
err := client.Do(ctx, client.B().Set().Key(key).Value(value).Ex(time.Hour).Build()).Error()
|
err := client.Do(ctx, client.B().Set().Key(key).Value(value).Ex(90*time.Minute).Build()).Error()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to set value in Valkey: %v", err)
|
return fmt.Errorf("failed to set value in Valkey: %v", err)
|
||||||
}
|
}
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
func GetValkeyValue(ctx context.Context, client valkey.Client, key string) (string, error) {
|
func GetValkeyValue(ctx context.Context, client valkey.Client, key string) (string, error) {
|
||||||
value, err := client.Do(ctx, client.B().Get().Key(key).Build()).ToString()
|
value, err := client.Do(ctx, client.B().Get().Key(key).Build()).ToString()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user