Compare commits
No commits in common. "f243fc68311edbf4f1b99c50c20c7f65ee60ec27" and "c4d680116a50443ce0813bd4ce23fdb990039b18" have entirely different histories.
f243fc6831
...
c4d680116a
|
@ -1,8 +1,6 @@
|
|||
package background
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
"time"
|
||||
|
||||
"git.fjla.uk/owlboard/timetable-mgr/cif"
|
||||
|
@ -11,16 +9,11 @@ import (
|
|||
"git.fjla.uk/owlboard/timetable-mgr/log"
|
||||
)
|
||||
|
||||
const frequency = 2 * time.Hour // Figure out a sensible frequency!
|
||||
const frequency = 600 * time.Hour // Figure out a sensible frequency!
|
||||
|
||||
// Starts a background ticker to run background tasks. Uses the frequency configured in the background/ticker.go file
|
||||
func InitTicker(cfg *helpers.Configuration, stop <-chan struct{}) {
|
||||
go runTicker(cfg, stop)
|
||||
|
||||
// Run goroutine logging ticker if runtime set to debug
|
||||
if helpers.Runtime == "debug" {
|
||||
go goroutineTicker(stop)
|
||||
}
|
||||
}
|
||||
|
||||
// Runs the ticker and handles tick events
|
||||
|
@ -39,28 +32,3 @@ func runTicker(cfg *helpers.Configuration, stop <-chan struct{}) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Starts a ticker that logs how many goroutines are running every two seconds
|
||||
func goroutineTicker(stop <-chan struct{}) {
|
||||
log.Msg.Warn("Starting goroutine Tracker ticker - DEBUG USE ONLY")
|
||||
ticker := time.NewTicker(1000 * time.Millisecond)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
case <-ticker.C:
|
||||
debugLog()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func debugLog() {
|
||||
var memStats runtime.MemStats
|
||||
runtime.ReadMemStats(&memStats)
|
||||
|
||||
goroutines := runtime.NumGoroutine()
|
||||
|
||||
fmt.Printf("\nNumber of goroutines: %d\n", goroutines)
|
||||
fmt.Printf("Heap Allocated memory: %2f MB\n\n", float64(memStats.HeapAlloc)/(1024*1024))
|
||||
}
|
||||
|
|
|
@ -46,7 +46,6 @@ func CheckCif(cfg *helpers.Configuration) {
|
|||
// Check how many days since last update, if more than 5, run full update, else run update
|
||||
daysSinceLastUpdate := howManyDaysAgo(metadata.LastUpdate)
|
||||
if daysSinceLastUpdate > 5 {
|
||||
log.Msg.Debug("Full Update Requested due to time since last update", zap.Int("daysSinceLastUpdate", daysSinceLastUpdate))
|
||||
log.Msg.Info("Full CIF download required")
|
||||
err := runCifFullDownload(cfg)
|
||||
if err != nil {
|
||||
|
@ -63,4 +62,6 @@ func CheckCif(cfg *helpers.Configuration) {
|
|||
if err != nil {
|
||||
log.Msg.Error("Unable to run CIF update", zap.Error(err))
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
|
|
@ -12,4 +12,4 @@ const fullUpdateUrl = "https://publicdatafeeds.networkrail.co.uk/ntrod/CifFileAu
|
|||
const dataAvailable = 6
|
||||
|
||||
// An object representing the Europe/London timezone
|
||||
var londonTimezone, _ = time.LoadLocation("Europe/London")
|
||||
var londonTimezone, err = time.LoadLocation("Europe/London")
|
||||
|
|
|
@ -21,13 +21,16 @@ func ConvertServiceType(input *upstreamApi.JsonScheduleV1, vstp bool) (*database
|
|||
PlanSpeed: parseSpeed(&input.ScheduleSegment.CifSpeed),
|
||||
ScheduleStartDate: ParseCifDate(&input.ScheduleStartDate, "start"),
|
||||
ScheduleEndDate: ParseCifDate(&input.ScheduleEndDate, "end"),
|
||||
FirstClass: hasFirstClass(&input.ScheduleSegment.CifTrainClass, &input.ScheduleSegment.SignallingId),
|
||||
FirstClass: hasFirstClass(&input.ScheduleSegment.CifTrainClass),
|
||||
Catering: hasCatering(&input.ScheduleSegment.CifCateringCode),
|
||||
Sleeper: hasSleeper(&input.ScheduleSegment.CifSleepers),
|
||||
DaysRun: parseDaysRun(&input.ScheduleDaysRun),
|
||||
Stops: parseStops(&input.ScheduleSegment.ScheduleLocation),
|
||||
// Remove CateringType from type definitions in Go and TS
|
||||
Sleeper: hasSleeper(&input.ScheduleSegment.CifSleepers),
|
||||
DaysRun: parseDaysRun(&input.ScheduleDaysRun),
|
||||
Stops: parseStops(&input.ScheduleSegment.ScheduleLocation),
|
||||
}
|
||||
|
||||
// New struct tags are not yet parsed. Ensure all struct tags are present.
|
||||
|
||||
return &output, nil
|
||||
}
|
||||
|
||||
|
@ -37,12 +40,14 @@ func parseSpeed(CIFSpeed *string) int32 {
|
|||
return 0
|
||||
}
|
||||
if *CIFSpeed == "" {
|
||||
log.Msg.Debug("Speed data not provided")
|
||||
return int32(0)
|
||||
}
|
||||
actualSpeed, exists := helpers.SpeedMap[*CIFSpeed]
|
||||
if !exists {
|
||||
actualSpeed = *CIFSpeed
|
||||
}
|
||||
log.Msg.Debug("Corrected Speed: " + actualSpeed)
|
||||
|
||||
speed, err := strconv.ParseInt(actualSpeed, 10, 32)
|
||||
if err != nil {
|
||||
|
@ -85,17 +90,10 @@ func isPublic(input *upstreamApi.CifScheduleLocation) bool {
|
|||
}
|
||||
|
||||
// Ascertains whether the service offers first class
|
||||
func hasFirstClass(input, signallingId *string) bool {
|
||||
if input == nil || signallingId == nil {
|
||||
func hasFirstClass(input *string) bool {
|
||||
if input == nil {
|
||||
return false
|
||||
}
|
||||
|
||||
// Handle non passenger headcodes and ensure first class is not shown as available
|
||||
firstChar := (*signallingId)[0]
|
||||
if firstChar == '3' || firstChar == '4' || firstChar == '5' || firstChar == '6' || firstChar == '7' || firstChar == '8' || firstChar == '0' {
|
||||
return false
|
||||
}
|
||||
|
||||
return *input != "S"
|
||||
}
|
||||
|
||||
|
|
|
@ -1,10 +1,8 @@
|
|||
package cif
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"git.fjla.uk/owlboard/go-types/pkg/database"
|
||||
"git.fjla.uk/owlboard/go-types/pkg/upstreamApi"
|
||||
)
|
||||
|
||||
|
@ -87,34 +85,23 @@ func TestIsPublic(t *testing.T) {
|
|||
|
||||
func TestHasFirstClass(t *testing.T) {
|
||||
testCases := []struct {
|
||||
input string
|
||||
headcode string
|
||||
expect bool
|
||||
input string
|
||||
expect bool
|
||||
}{
|
||||
{"", "1A00", true},
|
||||
{"B", "2A05", true},
|
||||
{"S", "1C99", false},
|
||||
{"", "3C23", false},
|
||||
{"", "5Q21", false},
|
||||
{"", "5D32", false},
|
||||
{"", "9O12", true},
|
||||
{"B", "9D32", true},
|
||||
{"", "7R43", false},
|
||||
{"B", "6Y77", false},
|
||||
{"", "8P98", false},
|
||||
{"S", "4O89", false},
|
||||
{"", "4E43", false},
|
||||
{"", true},
|
||||
{"B", true},
|
||||
{"S", false},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
result := hasFirstClass(&tc.input, &tc.headcode)
|
||||
result := hasFirstClass(&tc.input)
|
||||
|
||||
if result != tc.expect {
|
||||
t.Errorf("For %s & headcode %s, expected %t, but got %t", tc.input, tc.headcode, tc.expect, result)
|
||||
t.Errorf("For %s, expected %t, but got %t", tc.input, tc.expect, result)
|
||||
}
|
||||
}
|
||||
|
||||
nilResult := hasFirstClass(nil, nil)
|
||||
nilResult := hasFirstClass(nil)
|
||||
if nilResult {
|
||||
t.Errorf("hasFirstClass failed to handle nil pointer, expected %t, got %t", false, nilResult)
|
||||
}
|
||||
|
@ -172,51 +159,3 @@ func TestHasSleeper(t *testing.T) {
|
|||
t.Errorf("hasSleeper failed to handle nil pointer, expected %t, but got %t", false, nilResult)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseStops(t *testing.T) {
|
||||
testCases := []struct {
|
||||
input []upstreamApi.CifScheduleLocation
|
||||
expected []database.Stop
|
||||
}{
|
||||
{
|
||||
input: []upstreamApi.CifScheduleLocation{
|
||||
{
|
||||
LocationType: "LO",
|
||||
RecordIdentity: "Yes",
|
||||
TiplocCode: "BATHSPA",
|
||||
TiplocInstance: "0",
|
||||
Arrival: "0445",
|
||||
Departure: "0449",
|
||||
PublicDeparture: "0449",
|
||||
Pass: "",
|
||||
Platform: "1",
|
||||
Line: "DM",
|
||||
Path: "DM",
|
||||
EngineeringAllowance: "",
|
||||
PathingAllowance: "",
|
||||
PerformanceAllowance: "",
|
||||
}},
|
||||
expected: []database.Stop{
|
||||
{
|
||||
PublicDeparture: "0449",
|
||||
WttDeparture: "0449",
|
||||
PublicArrival: "",
|
||||
WttArrival: "0445",
|
||||
IsPublic: true,
|
||||
Tiploc: "BATHSPA",
|
||||
Pass: "",
|
||||
Platform: "1",
|
||||
ArrLine: "DM",
|
||||
DepLine: "DM",
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
result := parseStops(&tc.input)
|
||||
if !reflect.DeepEqual(result, tc.expected) {
|
||||
t.Errorf("Test case failed. Input: %v, Expected: %v, Got: %v", tc.input, tc.expected, result)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -44,13 +44,11 @@ func isSameToday(t time.Time) bool {
|
|||
|
||||
// Returns how many days ago `t` was compared to today
|
||||
func howManyDaysAgo(t time.Time) int {
|
||||
log.Msg.Debug("Calculating how many days ago", zap.Time("Input time", t))
|
||||
// Truncate both times to midnight in UTC timezone
|
||||
today := time.Now().UTC().Truncate(24 * time.Hour)
|
||||
input := t.UTC().Truncate(24 * time.Hour)
|
||||
today := time.Now().In(time.UTC).Truncate(24 * time.Hour)
|
||||
input := t.In(time.UTC).Truncate(24 * time.Hour)
|
||||
|
||||
diff := today.Sub(input)
|
||||
days := int(diff / (24 * time.Hour))
|
||||
days := int(diff.Hours() / 24)
|
||||
return days
|
||||
}
|
||||
|
||||
|
|
|
@ -24,11 +24,11 @@ func TestHowManyDaysAgo(t *testing.T) {
|
|||
input time.Time
|
||||
expected int
|
||||
}{
|
||||
{time.Now().In(time.UTC), 0}, // Today
|
||||
{time.Now().In(time.UTC).Add(-24 * time.Hour), 1}, // Yesterday
|
||||
{time.Now().In(time.UTC).Add(-48 * time.Hour), 2}, // Ereyesterday
|
||||
{time.Now().In(time.UTC).Add(24 * time.Hour), -1}, // Tomorrow
|
||||
{time.Now().In(time.UTC).Add(48 * time.Hour), -2}, // Overmorrow
|
||||
{time.Now(), 0}, // Today
|
||||
{time.Now().Add(-24 * time.Hour), 1}, // Yesterday
|
||||
{time.Now().Add(-48 * time.Hour), 2}, // Ereyesterday
|
||||
{time.Now().Add(24 * time.Hour), -1}, // Tomorrow
|
||||
{time.Now().Add(48 * time.Hour), -2}, // Overmorrow
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
|
|
|
@ -18,11 +18,9 @@ func processParsedCif(data *parsedData) error {
|
|||
for _, item := range data.sched {
|
||||
switch item.TransactionType {
|
||||
case "Delete":
|
||||
deleteItem := item // Create new variable to ensure repetition of pointers
|
||||
deleteTasks = append(deleteTasks, &deleteItem)
|
||||
deleteTasks = append(deleteTasks, &item)
|
||||
case "Create":
|
||||
createItem := item // Create new variable to ensure repetition of pointers
|
||||
createTasks = append(createTasks, &createItem)
|
||||
createTasks = append(createTasks, &item)
|
||||
default:
|
||||
log.Msg.Error("Unknown transaction type in CIF Schedule", zap.String("TransactionType", item.TransactionType))
|
||||
}
|
||||
|
|
|
@ -35,7 +35,6 @@ func TestGenerateMetadata(t *testing.T) {
|
|||
|
||||
if result == nil {
|
||||
t.Errorf("generateMetadata returned nil pointer")
|
||||
return // Static type checking likes this return to be here, even if it is redundant in reality.
|
||||
}
|
||||
|
||||
if result.Doctype != expected.Doctype {
|
||||
|
|
|
@ -32,7 +32,7 @@ func runCifFullDownload(cfg *helpers.Configuration) error {
|
|||
}
|
||||
|
||||
// Drop timetable collection
|
||||
dbAccess.DropCollection(dbAccess.TimetableCollection) // I should edit this to prevent removal of VSTP entries in the database.
|
||||
dbAccess.DropCollection(dbAccess.TimetableCollection)
|
||||
|
||||
// Process CIF file
|
||||
err = processParsedCif(parsed)
|
||||
|
@ -40,11 +40,7 @@ func runCifFullDownload(cfg *helpers.Configuration) error {
|
|||
log.Msg.Error("Error processing CIF data", zap.Error(err))
|
||||
}
|
||||
|
||||
newMeta := generateMetadata(&parsed.header)
|
||||
ok := dbAccess.PutCifMetadata(newMeta)
|
||||
if !ok {
|
||||
log.Msg.Warn("CIF Data updated, but metadata write failed")
|
||||
}
|
||||
// Generate & Write metadata
|
||||
|
||||
return nil
|
||||
}
|
||||
|
|
|
@ -18,10 +18,10 @@ const Doctype = "CifMetadata"
|
|||
// The type describing the CifMetadata 'type' in the database.
|
||||
// This type will be moved to owlboard/go-types
|
||||
type CifMetadata struct {
|
||||
Doctype string `bson:"type"`
|
||||
LastUpdate time.Time `bson:"lastUpdate"`
|
||||
LastTimestamp int64 `bson:"lastTimestamp"`
|
||||
LastSequence int64 `bson:"lastSequence"`
|
||||
Doctype string `json:"type"`
|
||||
LastUpdate time.Time `json:"lastUpdate"`
|
||||
LastTimestamp int64 `json:"lastTimestamp"`
|
||||
LastSequence int64 `json:"lastSequence"`
|
||||
}
|
||||
|
||||
// Fetches the CifMetadata from the database, returns nil if no metadata exists - before first initialisation for example.
|
||||
|
@ -39,24 +39,20 @@ func GetCifMetadata() (*CifMetadata, error) {
|
|||
return nil, err
|
||||
}
|
||||
|
||||
log.Msg.Debug("Fetched CIF Metadata from database", zap.Any("Metadata", result))
|
||||
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
// Uses upsert to Insert/Update the CifMetadata in the database
|
||||
func PutCifMetadata(metadata *CifMetadata) bool {
|
||||
func PutCifMetadata(metadata CifMetadata) bool {
|
||||
database := MongoClient.Database(databaseName)
|
||||
collection := database.Collection(metaCollection)
|
||||
options := options.Update().SetUpsert(true)
|
||||
filter := bson.M{"type": Doctype}
|
||||
update := bson.M{
|
||||
"$set": bson.M{
|
||||
"type": Doctype,
|
||||
"lastUpdate": metadata.LastUpdate,
|
||||
"lastTimestamp": metadata.LastTimestamp,
|
||||
"lastSequence": metadata.LastSequence,
|
||||
},
|
||||
"type": Doctype,
|
||||
"LastUpdate": metadata.LastUpdate,
|
||||
"LastTimestamp": metadata.LastTimestamp,
|
||||
"LastSequence": metadata.LastSequence,
|
||||
}
|
||||
|
||||
_, err := collection.UpdateOne(context.Background(), filter, update, options)
|
||||
|
@ -65,21 +61,14 @@ func PutCifMetadata(metadata *CifMetadata) bool {
|
|||
log.Msg.Error("Error updating CIF Metadata", zap.Error(err))
|
||||
return false
|
||||
}
|
||||
|
||||
log.Msg.Info("New CIF Metadata written", zap.Time("Update time", metadata.LastUpdate))
|
||||
return true
|
||||
}
|
||||
|
||||
// Handles 'Delete' tasks from CIF Schedule updates, accepts DeleteQuery types and batches deletions.
|
||||
func DeleteCifEntries(deletions []database.DeleteQuery) error {
|
||||
// Skip if deletions is empty
|
||||
if len(deletions) == 0 {
|
||||
log.Msg.Info("No deletions required")
|
||||
return nil
|
||||
}
|
||||
collection := MongoClient.Database(databaseName).Collection(timetableCollection)
|
||||
|
||||
// Prepare deletion tasks
|
||||
collection := MongoClient.Database(databaseName).Collection(timetableCollection)
|
||||
bulkDeletions := make([]mongo.WriteModel, 0, len(deletions))
|
||||
|
||||
for _, deleteQuery := range deletions {
|
||||
|
@ -106,12 +95,6 @@ func DeleteCifEntries(deletions []database.DeleteQuery) error {
|
|||
|
||||
// Handles 'Create' tasks for CIF Schedule updates, accepts Service structs and batches their creation.
|
||||
func CreateCifEntries(schedules []database.Service) error {
|
||||
// Skip if deletions is empty
|
||||
if len(schedules) == 0 {
|
||||
log.Msg.Info("No creations required")
|
||||
return nil
|
||||
}
|
||||
|
||||
collection := MongoClient.Database(databaseName).Collection(timetableCollection)
|
||||
|
||||
models := make([]mongo.WriteModel, 0, len(schedules))
|
||||
|
|
2
go.mod
2
go.mod
|
@ -3,7 +3,7 @@ module git.fjla.uk/owlboard/timetable-mgr
|
|||
go 1.21
|
||||
|
||||
require (
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240408193146-4719be9c13eb
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240407202712-e58d7d1d9aa9
|
||||
github.com/go-stomp/stomp/v3 v3.0.5
|
||||
go.mongodb.org/mongo-driver v1.12.0
|
||||
go.uber.org/zap v1.24.0
|
||||
|
|
4
go.sum
4
go.sum
|
@ -8,10 +8,6 @@ git.fjla.uk/owlboard/go-types v0.0.0-20240405194933-7dafca950460 h1:39vcejk0MzZI
|
|||
git.fjla.uk/owlboard/go-types v0.0.0-20240405194933-7dafca950460/go.mod h1:kG+BX9UF+yJaAVnln/QSKlTdrtKRRReezMeSk1ZLMzY=
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240407202712-e58d7d1d9aa9 h1:aNxMYEsbBkFx9b9J8zR1m08KL3JG9b0rPHszryKm5uk=
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240407202712-e58d7d1d9aa9/go.mod h1:kG+BX9UF+yJaAVnln/QSKlTdrtKRRReezMeSk1ZLMzY=
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240408150352-8ba2a306a580 h1:bEaC1JfqiSSJH65iP/NXMyBo85JMB41VBkiJdWbnHYM=
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240408150352-8ba2a306a580/go.mod h1:kG+BX9UF+yJaAVnln/QSKlTdrtKRRReezMeSk1ZLMzY=
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240408193146-4719be9c13eb h1:aLd0nzuU13hxycz9F4Z4PVq5dp/TxuzywPGZTJXbnq0=
|
||||
git.fjla.uk/owlboard/go-types v0.0.0-20240408193146-4719be9c13eb/go.mod h1:kG+BX9UF+yJaAVnln/QSKlTdrtKRRReezMeSk1ZLMzY=
|
||||
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
|
||||
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
|
|
|
@ -19,7 +19,7 @@ func init() {
|
|||
|
||||
// Determine the log level based on the runtime mode
|
||||
logLevel := zapcore.DebugLevel
|
||||
if helpers.Runtime != "debug" {
|
||||
if helpers.Runtime == "production" {
|
||||
logLevel = zapcore.InfoLevel
|
||||
}
|
||||
|
||||
|
|
11
main.go
11
main.go
|
@ -9,7 +9,6 @@ import (
|
|||
_ "time/tzdata"
|
||||
|
||||
"git.fjla.uk/owlboard/timetable-mgr/background"
|
||||
"git.fjla.uk/owlboard/timetable-mgr/cif"
|
||||
"git.fjla.uk/owlboard/timetable-mgr/corpus"
|
||||
"git.fjla.uk/owlboard/timetable-mgr/dbAccess"
|
||||
"git.fjla.uk/owlboard/timetable-mgr/helpers"
|
||||
|
@ -35,15 +34,13 @@ func main() {
|
|||
// Initialise a `stop` channel to signal goroutines to cleanup
|
||||
stop := make(chan struct{})
|
||||
|
||||
// Start CIF Task Ticker
|
||||
background.InitTicker(cfg, stop)
|
||||
|
||||
// Handle signals from the OS
|
||||
go handleSignals(cfg, stop)
|
||||
|
||||
// Manually call CIF and CORPUS checks to ensure that they are
|
||||
// not delayed until the first ticker event.
|
||||
go cif.CheckCif(cfg)
|
||||
// Start CIF Task Ticker
|
||||
background.InitTicker(cfg, stop)
|
||||
|
||||
// Test CORPUS Fetching
|
||||
go corpus.CheckCorpus(cfg)
|
||||
|
||||
if cfg.VstpOn {
|
||||
|
|
Loading…
Reference in New Issue