Added a background trigger to periodically start msgInLog processing. Also, some clean-ups.

Author: numbleroot

comm/connection.go

@@ -41,10 +41,9 @@ func ReliableConnect(remoteAddr string, tlsConfig *tls.Config, retry int) (*tls.
 	return c, nil
 }
 
-// ReliableSend attempts to transmit a message inbetween
-// pluto nodes. If the first attempt fails, the node will
-// try to reconnect and resend the message until successfully
-// transmitted.
+// ReliableSend attempts to transmit a message between pluto
+// nodes. If the first attempt fails, the node will try to
+// reconnect and resend the message until successfully transmitted.
 func ReliableSend(conn *tls.Conn, text string, remoteAddr string, tlsConfig *tls.Config, timeout int, retry int) error {
 
 	// TODO: Make this routine first check for closed

comm/receiver.go

@@ -11,6 +11,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"io/ioutil"
 )
@@ -123,6 +124,10 @@ func InitReceiver(name string, logFilePath string, vclockLogPath string, socket
 	// initial run to check if log file contains elements.
 	recv.msgInLog <- struct{}{}
 
+	// Start triggering msgInLog events periodically.
+	recv.wg.Add(1)
+	go recv.TriggerMsgApplier()
+
 	// Accept incoming messages in background.
 	recv.wg.Add(1)
 	go recv.AcceptIncMsgs()
@@ -223,6 +228,7 @@ func (recv *Receiver) Shutdown(downRecv chan struct{}) {
 	recv.shutdown <- struct{}{}
 	recv.shutdown <- struct{}{}
 	recv.shutdown <- struct{}{}
+	recv.shutdown <- struct{}{}
 
 	// Close involved channels.
 	close(recv.incVClock)
@@ -336,6 +342,45 @@ func (recv *Receiver) AcceptIncMsgs() error {
 	}
 }
 
+// TriggerMsgApplier starts a timer that triggers
+// an msgInLog event when duration elapsed. Supposed
+// to routinely poke the ApplyStoredMsgs into checking
+// for unprocessed messages in log.
+func (recv *Receiver) TriggerMsgApplier() {
+
+	// Specify duration to wait between triggers.
+	triggerD := 2 * time.Second
+
+	// Create a timer that waits for one second
+	// to elapse and then fires.
+	triggerT := time.NewTimer(triggerD)
+
+	for {
+
+		select {
+
+		// Check if a shutdown signal was sent.
+		case <-recv.shutdown:
+
+			// Call done handler of wait group for this
+			// routine on exiting this function.
+			defer recv.wg.Done()
+			return
+
+		case <-triggerT.C:
+
+			// If buffered channel indicating an arrived
+			// msg is not full yet, make it full.
+			if len(recv.msgInLog) < 1 {
+				recv.msgInLog <- struct{}{}
+			}
+
+			// Renew timer.
+			triggerT.Reset(triggerD)
+		}
+	}
+}
+
 // StoreIncMsgs takes received message string and saves
 // it into incoming CRDT message log file.
 func (recv *Receiver) StoreIncMsgs(conn net.Conn) {
@@ -358,9 +403,9 @@ func (recv *Receiver) StoreIncMsgs(conn net.Conn) {
 			if err.Error() == "EOF" {
 				log.Printf("[comm.StoreIncMsgs] Reading from closed connection. Ignoring.\n")
 				return
-			} else {
-				log.Fatalf("[comm.StoreIncMsgs] Error while reading sync message: %s\n", err.Error())
 			}
+
+			log.Fatalf("[comm.StoreIncMsgs] Error while reading sync message: %s\n", err.Error())
 		}
 
 		// Remove trailing characters denoting line end.

imap/distributor-commands.go

@@ -39,8 +39,8 @@ func (distr *Distributor) Capability(c *Connection, req *Request) bool {
 		return false
 	}
 
-	// TODO: Change returned capabilites based on IMAP state of
-	//       connection, e.g. more capabilites if authenticated.
+	// TODO: Change returned capabilities based on IMAP state of
+	//       connection, e.g. more capabilities if authenticated.
 
 	return true
 }