Merge branch '311_dfssd_log_sorting' into 'master'

311 dfssd log sorting

This is quick work, I'll take any insight.
I didn't do it, but an integration test using mutex can be nice.

See merge request !42

dfssd/dfssd_test.go

@@ -24,4 +24,11 @@ func TestServerAndClient(t *testing.T) {
 		// this one fails silently, so you can't really test it
 		dapi.DLog("This is a log message from a client")
 	}()
+
+	// Start another client
+	go func() {
+		defer dapi.DClose()
+		// this one fails silently, so you can't really test it
+		dapi.DLog("This is a log message from another client")
+	}()
 }

dfssd/server.go

 package main
 
 import (
-	api "dfss/dfssd/api"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/grpclog"
 	"log"
 	"net"
+
+	api "dfss/dfssd/api"
 )
 
 type server struct{}
@@ -15,8 +16,7 @@ type server struct{}
 //
 // Handle incoming log messages
 func (s *server) SendLog(ctx context.Context, in *api.Log) (*api.Ack, error) {
-	// TODO send message to log management
-	log.Printf("[%d] %s:: %s", in.Timestamp, in.Identifier, in.Log)
+	addMessage(in)
 	return &api.Ack{}, nil
 }
 
@@ -30,6 +30,9 @@ func listen(addrPort string) error {
 	}
 	log.Printf("Server listening on %s", addrPort)
 
+	// log display manager
+	go displayHandler()
+
 	// bootstrap gRPC service !
 	grpcServer := grpc.NewServer()
 	api.RegisterDemonstratorServer(grpcServer, &server{})

dfssd/storage.go

+package main
+
+import (
+	"log"
+	"sort"
+	"sync"
+	"time"
+
+	"dfss/dfssd/api"
+)
+
+// NOTE: buffers are managed with slices since lists are pretty much not used in go
+// @see https://github.com/golang/go/wiki/SliceTricks
+
+var in []*api.Log // incoming msg buffer
+var inMutex = &sync.Mutex{}
+
+// addMessage to storage
+func addMessage(msg *api.Log) {
+	inMutex.Lock()
+	in = append(in, msg)
+	inMutex.Unlock()
+}
+
+// display logs that are more than since (ms) old
+func display(since int64) {
+	var out []*api.Log      // sorted messages to display
+	var recycled []*api.Log // messages too recent to be displayed
+
+	present := time.Now().UnixNano()
+
+	inMutex.Lock()
+
+	for _, v := range in {
+		if present-(*v).Timestamp > 1000000*since {
+			out = append(out, v)
+		} else {
+			recycled = append(recycled, v)
+		}
+	}
+
+	in = recycled
+	inMutex.Unlock()
+
+	sort.Sort(ByTimestamp(out))
+
+	for _, v := range out {
+		log.Printf("[%d] %s:: %s", v.Timestamp, v.Identifier, v.Log)
+	}
+}
+
+// refresh every second
+func displayHandler() {
+	ticker := time.NewTicker(time.Second)
+	for range ticker.C {
+		display(1000)
+	}
+}
+
+// ByTimestamp sorting interface
+type ByTimestamp []*api.Log
+
+func (l ByTimestamp) Len() int {
+	return len(l)
+}
+func (l ByTimestamp) Swap(i, j int) {
+	l[i], l[j] = l[j], l[i]
+}
+func (l ByTimestamp) Less(i, j int) bool {
+	return (*l[i]).Timestamp < (*l[j]).Timestamp
+}