diff options
| -rw-r--r-- | cache.go | 90 | ||||
| -rw-r--r-- | init.go | 16 |
2 files changed, 68 insertions, 38 deletions
diff --git a/cache.go b/cache.go index 253de98..be3d7d3 100644 --- a/cache.go +++ b/cache.go @@ -10,18 +10,20 @@ import ( "github.com/syndtr/goleveldb/leveldb" ) -// checks if it's time to refresh the cache or not +// Checks whether it's time to refresh +// the cache. func checkCacheTime() bool { return time.Since(confObj.lastCache) > confObj.cacheInterval } -// checks if it's time to push the cache to the database +// Checks whether it's time to push +// the cache to the database func checkDBtime() bool { return time.Since(confObj.lastPush) > confObj.dbInterval } -// launched by init as a goroutine to constantly watch -// for the update interval to pass +// Launched by init as a goroutine to constantly watch +// for the update interval to pass. func cacheAndPush() { for { if checkCacheTime() { @@ -35,9 +37,11 @@ func cacheAndPush() { } } -// refreshes the cache +// Refreshes the cache. func refreshCache() { + // Iterate over the registry and + // update each individual user. for k := range twtxtCache.Reg { err := twtxtCache.UpdateUser(k) if err != nil { @@ -46,6 +50,9 @@ func refreshCache() { } } + // Re-scrape all the remote registries + // to see if they have any new users + // to add locally. for _, v := range remoteRegistries.List { err := twtxtCache.ScrapeRemoteRegistry(v) if err != nil { @@ -57,40 +64,46 @@ func refreshCache() { confObj.mu.Unlock() } -// pushes the registry's cache data to a local -// database for safe keeping +// Pushes the registry's cache data to a local +// database for safe keeping. func pushDatabase() error { + // Acquire the database from the aether. + // goleveldb is concurrency-safe, so we + // can immediately push it back into the + // channel for other functions to use. db := <-dbChan - twtxtCache.Mu.RLock() + dbChan <- db - // create a batch write job so it can + // Create a batch write job so it can // be done at one time rather than - // per value + // per entry. + twtxtCache.Mu.RLock() var dbBasket *leveldb.Batch for k, v := range twtxtCache.Reg { - dbBasket.Put([]byte(k+".Nick"), []byte(v.Nick)) - dbBasket.Put([]byte(k+".URL"), []byte(v.URL)) - dbBasket.Put([]byte(k+".IP"), []byte(v.IP)) - dbBasket.Put([]byte(k+".Date"), []byte(v.Date)) + dbBasket.Put([]byte(k+"*Nick"), []byte(v.Nick)) + dbBasket.Put([]byte(k+"*URL"), []byte(v.URL)) + dbBasket.Put([]byte(k+"*IP"), []byte(v.IP)) + dbBasket.Put([]byte(k+"*Date"), []byte(v.Date)) for i, e := range v.Status { - dbBasket.Put([]byte(k+".Status."+i.String()), []byte(e)) + rfc := i.Format(time.RFC3339) + dbBasket.Put([]byte(k+"*Status*"+rfc), []byte(e)) } } + twtxtCache.Mu.RUnlock() - // save our list of remote registries to scrape + // Save our list of remote registries to scrape. for k, v := range remoteRegistries.List { - dbBasket.Put([]byte("remote."+string(k)), []byte(v)) + dbBasket.Put([]byte("remote*"+string(k)), []byte(v)) } - // execute the batch job + // Execute the batch job. if err := db.Write(dbBasket, nil); err != nil { return err } - twtxtCache.Mu.RUnlock() - dbChan <- db - - // update the last push time + // Update the last push time for + // our timer/watch function to + // reference. confObj.mu.Lock() confObj.lastPush = time.Now() confObj.mu.Unlock() @@ -98,29 +111,42 @@ func pushDatabase() error { return nil } -// pulls registry data from the DB on startup +// Pulls registry data from the DB on startup. +// Iterates over the database one entry at a time. func pullDatabase() { + // Acquire the database from the aether. + // goleveldb is concurrency-safe, so we + // can immediately push it back into the + // channel for other functions to use. db := <-dbChan + dbChan <- db iter := db.NewIterator(nil, nil) + // Read the database key-by-key for iter.Next() { key := iter.Key() val := iter.Value() - split := strings.Split(string(key), ".") + split := strings.Split(string(key), "*") urls := string(split[0]) field := string(split[1]) - data := registry.NewUserData() + // Start with an empty Data struct. If + // there's already one in the cache, pull + // it and use it instead. + data := registry.NewUserData() twtxtCache.Mu.RLock() if _, ok := twtxtCache.Reg[urls]; ok { data = twtxtCache.Reg[urls] } twtxtCache.Mu.RUnlock() - ref := reflect.ValueOf(data).Elem() + // Use reflection to find the right field + // in the Data struct. Once found, assign + // the value and break so the DB iteration + // can continue. if field != "Status" && urls != "remote" { for i := 0; i < ref.NumField(); i++ { @@ -132,6 +158,9 @@ func pullDatabase() { } } else if field == "Status" && urls != "remote" { + // If we're looking at a Status entry in the DB, + // parse the time then add it to the TimeMap under + // data.Status thetime, err := time.Parse("RFC3339", split[2]) if err != nil { log.Printf("%v\n", err) @@ -139,11 +168,18 @@ func pullDatabase() { data.Status[thetime] = string(val) } else { + // The third and final possibility is + // if we've come across an entry for + // a remote twtxt registry to scrape. + // If so, add it to our list. remoteRegistries.Mu.Lock() remoteRegistries.List = append(remoteRegistries.List, string(val)) remoteRegistries.Mu.Unlock() + continue } + // Push the data struct (back) into + // the cache. twtxtCache.Mu.Lock() twtxtCache.Reg[urls] = data twtxtCache.Mu.Unlock() @@ -154,6 +190,4 @@ func pullDatabase() { if err != nil { log.Printf("Error while pulling DB into registry cache: %v\n", err) } - - dbChan <- db } diff --git a/init.go b/init.go index 6896a1a..639a78b 100644 --- a/init.go +++ b/init.go @@ -33,11 +33,6 @@ var closelog = make(chan bool, 1) // initialization var dbChan = make(chan *leveldb.DB, 1) -// provides access to the database so it -// can be closed outside of the init function's -// scope. -var dbCloseChan = make(chan *leveldb.DB, 1) - // templates var tmpls *template.Template @@ -54,7 +49,6 @@ func init() { initLogging() tmpls = initTemplates() initDatabase() - go cacheAndPush() watchForInterrupt() } @@ -218,17 +212,19 @@ func initTemplates() *template.Template { return template.Must(template.ParseFiles("assets/tmpl/index.html")) } -// Pull DB data into cache, if available +// Pull DB data into cache, if available. func initDatabase() { db, err := leveldb.OpenFile(confObj.dbPath, nil) if err != nil { log.Fatalf("%v\n", err) } + // Send the database reference into + // the aether. dbChan <- db - dbCloseChan <- db pullDatabase() + go cacheAndPush() } // Watch for SIGINT aka ^C @@ -245,7 +241,7 @@ func watchForInterrupt() { // Close the database cleanly log.Printf("Closing database connection to %v...\n", confObj.dbPath) - db := <-dbCloseChan + db := <-dbChan if err := db.Close(); err != nil { log.Printf("%v\n", err) } @@ -256,7 +252,7 @@ func watchForInterrupt() { } confObj.mu.RUnlock() - close(dbCloseChan) + close(dbChan) close(closelog) // Let everything catch up |