extract code in helper function and uniform the code

reintroduce the closing of input channel (it's required)

Author: umbynos

bufferflow_timed.go

@@ -31,33 +31,32 @@ func NewBufferflowTimed(port string, output chan []byte) *BufferflowTimed {
 
 func (b *BufferflowTimed) Init() {
 	log.Println("Initting timed buffer flow (output once every 16ms)")
+	go b.consumeInput()
+}
 
-	go func() {
-	Loop:
-		for {
-			select {
-			case data := <-b.input:
-				b.bufferedOutput = b.bufferedOutput + data
-				b.sPort = b.port
-			case <-b.ticker.C:
-				if b.bufferedOutput != "" {
-					m := SpPortMessage{b.sPort, b.bufferedOutput}
-					buf, _ := json.Marshal(m)
-					// data is now encoded in base64 format
-					// need a decoder on the other side
-					b.output <- []byte(buf)
-					b.bufferedOutput = ""
-					b.sPort = ""
-				}
-			case <-b.done:
-				break Loop
+func (b *BufferflowTimed) consumeInput() {
+Loop:
+	for {
+		select {
+		case data := <-b.input: // use the buffer and append data to it
+			b.bufferedOutput = b.bufferedOutput + data
+			b.sPort = b.port
+		case <-b.ticker.C: // after 16ms send the buffered output message
+			if b.bufferedOutput != "" {
+				m := SpPortMessage{b.sPort, b.bufferedOutput}
+				buf, _ := json.Marshal(m)
+				// data is now encoded in base64 format
+				// need a decoder on the other side
+				b.output <- buf
+				// reset the buffer and the port
+				b.bufferedOutput = ""
+				b.sPort = ""
 			}
+		case <-b.done:
+			break Loop //this is required, a simple break statement would only exit the innermost switch statement
 		}
-
-		close(b.input)
-
-	}()
-
+	}
+	close(b.input)
 }
 
 func (b *BufferflowTimed) BlockUntilReady(cmd string, id string) (bool, bool) {

bufferflow_timedbinary.go

@@ -31,28 +31,32 @@ func NewBufferflowTimedBinary(port string, output chan []byte) *BufferflowTimedB
 
 func (b *BufferflowTimedBinary) Init() {
 	log.Println("Initting timed buffer binary flow (output once every 16ms)")
-	go func() {
-	Loop:
-		for {
-			select {
-			case data := <-b.input:
-				b.bufferedOutputBinary = append(b.bufferedOutputBinary, data...)
-				b.sPortBinary = b.port
-			case <-b.ticker.C:
-				if b.bufferedOutputBinary != nil {
-					m := SpPortMessageRaw{b.sPortBinary, b.bufferedOutputBinary}
-					buf, _ := json.Marshal(m)
-					b.output <- buf
-					b.bufferedOutputBinary = nil
-					b.sPortBinary = ""
-				}
-			case <-b.done:
-				break Loop
+	go b.consumeInput()
+}
+
+func (b *BufferflowTimedBinary) consumeInput() {
+Loop:
+	for {
+		select {
+		case data := <-b.input: // use the buffer and append data to it
+			b.bufferedOutputBinary = append(b.bufferedOutputBinary, data...)
+			b.sPortBinary = b.port
+		case <-b.ticker.C: // after 16ms send the buffered output message
+			if b.bufferedOutputBinary != nil {
+				m := SpPortMessageRaw{b.sPortBinary, b.bufferedOutputBinary}
+				buf, _ := json.Marshal(m)
+				// data is now encoded in base64 format
+				// need a decoder on the other side
+				b.output <- buf
+				// reset the buffer and the port
+				b.bufferedOutputBinary = nil
+				b.sPortBinary = ""
 			}
+		case <-b.done:
+			break Loop //this is required, a simple break statement would only exit the innermost switch statement
 		}
-
-		close(b.input)
-	}()
+	}
+	close(b.input)
 }
 
 func (b *BufferflowTimedBinary) BlockUntilReady(cmd string, id string) (bool, bool) {
@@ -70,5 +74,6 @@ func (b *BufferflowTimedBinary) IsBufferGloballySendingBackIncomingData() bool {
 
 func (b *BufferflowTimedBinary) Close() {
 	b.ticker.Stop()
+	b.done <- true
 	close(b.input)
 }

bufferflow_timedraw.go

@@ -31,31 +31,32 @@ func NewBufferflowTimedRaw(port string, output chan []byte) *BufferflowTimedRaw
 
 func (b *BufferflowTimedRaw) Init() {
 	log.Println("Initting timed buffer raw flow (output once every 16ms)")
+	go b.consumeInput()
+}
 
-	go func() {
-	Loop:
-		for {
-			select {
-			case data := <-b.input:
-				b.bufferedOutputRaw = append(b.bufferedOutputRaw, []byte(data)...)
-				b.sPortRaw = b.port
-			case <-b.ticker.C:
-				if b.bufferedOutputRaw != nil {
-					m := SpPortMessageRaw{b.sPortRaw, b.bufferedOutputRaw}
-					buf, _ := json.Marshal(m)
-					// data is now encoded in base64 format
-					// need a decoder on the other side
-					b.output <- []byte(buf)
-					b.bufferedOutputRaw = nil
-					b.sPortRaw = ""
-				}
-			case <-b.done:
-				break Loop
+func (b *BufferflowTimedRaw) consumeInput() {
+Loop:
+	for {
+		select {
+		case data := <-b.input: // use the buffer and append data to it
+			b.bufferedOutputRaw = append(b.bufferedOutputRaw, []byte(data)...)
+			b.sPortRaw = b.port
+		case <-b.ticker.C: // after 16ms send the buffered output message
+			if b.bufferedOutputRaw != nil {
+				m := SpPortMessageRaw{b.sPortRaw, b.bufferedOutputRaw}
+				buf, _ := json.Marshal(m)
+				// data is now encoded in base64 format
+				// need a decoder on the other side
+				b.output <- buf
+				// reset the buffer and the port
+				b.bufferedOutputRaw = nil
+				b.sPortRaw = ""
 			}
+		case <-b.done:
+			break Loop //this is required, a simple break statement would only exit the innermost switch statement
 		}
-
-		close(b.input)
-	}()
+	}
+	close(b.input)
 }
 
 func (b *BufferflowTimedRaw) BlockUntilReady(cmd string, id string) (bool, bool) {
@@ -73,5 +74,6 @@ func (b *BufferflowTimedRaw) IsBufferGloballySendingBackIncomingData() bool {
 
 func (b *BufferflowTimedRaw) Close() {
 	b.ticker.Stop()
+	b.done <- true
 	close(b.input)
 }