oct design v 1.4.3 - clean up contd

Author: rapidclock

octopus/pipe_spl_ingest.go

@@ -1,11 +1,7 @@
 package octopus
 
-import (
-	"time"
-)
-
 func (o *octopus) makeIngestPipe(inChSet *ingestPipeChSet, opChSet *NodeChSet) {
-	go channelConnector(inChSet, opChSet, o.timeToQuit, o.masterQuitCh)
+	go channelConnector(inChSet, opChSet)
 	go setupStringIngestPipe(inChSet, opChSet, o.masterQuitCh)
 }
 
@@ -26,36 +22,11 @@ func setupStringIngestPipe(inChSet *ingestPipeChSet, nodeOpChSet *NodeChSet,
 	}
 }
 
-func channelConnector(inChSet *ingestPipeChSet, opChSet *NodeChSet,
-	timeOut time.Duration, masterQuitCh chan int) {
-	// timeOutTimer := time.NewTimer(timeOut)
+func channelConnector(inChSet *ingestPipeChSet, opChSet *NodeChSet) {
 	for {
-		// timeOutCh = time.After(timeOut * time.Second)
-		// timeOutCh = time.NewTimer(timeOut)
 		select {
 		case node := <-inChSet.NodeCh:
 			opChSet.NodeCh <- node
-			// if !timeOutTimer.Stop() {
-			// 	<-timeOutTimer.C
-			// }
-			// log.Println("abc")
-			// timeOutTimer.Reset(timeOut)
-			// case i := <-inChSet.QuitCh:
-			// 	{
-			// 		fmt.Println("Quit Received on Ingest Channel")
-			// opChSet.QuitCh <- i
-			// 		masterQuitCh <- i
-			// 		if !timeOutTimer.Stop() {
-			// 			<-timeOutTimer.C
-			// 		}
-			// 		timeOutTimer.Reset(timeOut)
-			// 	}
-			// case <-timeOutTimer.C:
-			// 	fmt.Println("Timeout Triggered in Ingest Channel")
-			// 	opChSet.QuitCh <- 1
-			// 	masterQuitCh <- 1
-			// 	return
-
 		}
 	}
 }

octopus/stdpipefunc.go

@@ -9,16 +9,8 @@ type stdFunc func(*Node, *NodeChSet)
 func stdLinearNodeFunc(stdFn stdFunc, outChSet *NodeChSet) *NodeChSet {
 	listenCh := make(chan *Node)
 	listenQuitCh := make(chan int, 1)
-	listenChSet := &NodeChSet{
-		NodeCh: listenCh,
-		StdChannels: &StdChannels{
-			QuitCh: listenQuitCh,
-		},
-	}
-	// listenChSet := MakeNodeChSet(listenCh, listenQuitCh)
+	listenChSet := MakeNodeChSet(listenCh, listenQuitCh)
 	go func() {
-		// defer close(listenCh)
-		// defer close(listenQuitCh)
 		for {
 			select {
 			case node := <-listenCh: