Merge pull request #2 from RangelReale/protocol-fix

* Allows lines not ending in \n, as it is not required by the protocol

statsd/aggregator.go

@@ -21,6 +21,14 @@ type MetricSender interface {
 	SendMetrics(MetricMap) error
 }
 
+// The MetricSenderFunc type is an adapter to allow the use of ordinary functions as metric senders
+type MetricSenderFunc func(MetricMap) error
+
+// SendMetrics calls f(m)
+func (f MetricSenderFunc) SendMetrics(m MetricMap) error {
+	return f(m)
+}
+
 // MetricAggregator is an object that aggregates statsd metrics.
 // The function NewMetricAggregator should be used to create the objects.
 //

statsd/receiver.go

@@ -69,25 +69,33 @@ func (r *MetricReceiver) Receive(c net.PacketConn) error {
 func (srv *MetricReceiver) handleMessage(addr net.Addr, msg []byte) {
 	buf := bytes.NewBuffer(msg)
 	for {
-		line, err := buf.ReadBytes('\n')
-		if err == io.EOF {
-			break
-		}
-		if err != nil {
-			log.Printf("error reading message from %s: %s", addr, err)
+		line, readerr := buf.ReadBytes('\n')
+
+		// protocol does not require line to end in \n, if EOF use received line if valid
+		if readerr != nil && readerr != io.EOF {
+			log.Printf("error reading message from %s: %s", addr, readerr)
 			return
+		} else if readerr != io.EOF {
+			// remove newline, only if not EOF
+			if len(line) > 0 {
+				line = line[:len(line)-1]
+			}
 		}
 
-		lineLength := len(line)
 		// Only process lines with more than one character
-		if lineLength > 1 {
-			metric, err := parseLine(line[:lineLength-1])
+		if len(line) > 1 {
+			metric, err := parseLine(line)
 			if err != nil {
-				log.Println("error parsing line %q from %s: %s", line, addr, err)
+				log.Printf("error parsing line %q from %s: %s", line, addr, err)
 				continue
 			}
 			go srv.Handler.HandleMetric(metric)
 		}
+
+		if readerr == io.EOF {
+			// if was EOF, finished handling
+			return
+		}
 	}
 }