design octopus v1.2

Author: rapidclock

octopus/core.go

@@ -1,19 +1,33 @@
 package octopus
 
 import (
+	"fmt"
 	"log"
 	"time"
 )
 
+func (o *octopus) setupOctopus() {
+	o.setupValidProtocolMap()
+	o.setupTimeToQuit()
+}
+
 func (o *octopus) setupValidProtocolMap() {
 	o.isValidProtocol = make(map[string]bool)
 	for _, protocol := range o.ValidProtocols {
 		o.isValidProtocol[protocol] = true
 	}
 }
 
+func (o *octopus) setupTimeToQuit() {
+	if o.TimeToQuit > 0 {
+		o.timeToQuit = time.Duration(o.TimeToQuit)
+	} else {
+		log.Fatalln("TimeToQuit is not greater than 0")
+	}
+}
+
 func (o *octopus) SetupSystem() {
-	o.setupValidProtocolMap()
+	o.setupOctopus()
 
 	ingestCh := make(chan *Node)
 	ingestQuitCh := make(chan int, 1)
@@ -39,7 +53,7 @@ func (o *octopus) SetupSystem() {
 	o.makeIngestPipe(inPipeChSet, linkAbsChSet)
 
 	o.inpUrlStrChan = ingestStrCh
-	o.masterQuitCh = ingestQuitCh
+	o.masterQuitCh = make(chan int, 1)
 	o.isReady = true
 }
 
@@ -56,9 +70,9 @@ func (o *octopus) BeginCrawling(baseUrlStr string) {
 			{
 				o.inpUrlStrChan <- urlStr
 			}
-		case <-time.After(10 * time.Second):
+		case <-o.masterQuitCh:
 			{
-				o.masterQuitCh <- 1
+				fmt.Println("Master Kill Switch Activated")
 				return
 			}
 		}

octopus/modelfactory.go

@@ -3,9 +3,10 @@ package octopus
 import "sync"
 
 const (
-	defaultMaxDepth int64 = 2
-	anchorTag             = "a"
-	anchorAttrb           = "href"
+	defaultMaxDepth   int64 = 2
+	anchorTag               = "a"
+	anchorAttrb             = "href"
+	defaultTimeToQuit       = 5
 )
 
 // NewWithDefaultOptions - Create an Instance of the Octopus with the default CrawlOptions.
@@ -49,6 +50,7 @@ func GetDefaultCrawlOptions() *CrawlOptions {
 		IncludeBody:        true,
 		OpAdapter:          nil,
 		ValidProtocols:     []string{"http", "https"},
+		TimeToQuit:         defaultTimeToQuit,
 	}
 }
 

octopus/models.go

@@ -3,6 +3,7 @@ package octopus
 import (
 	"io"
 	"sync"
+	"time"
 )
 
 // octopus is a concurrent web crawler.
@@ -15,6 +16,7 @@ type octopus struct {
 	isReady         bool
 	adapterChSet    *NodeChSet
 	isValidProtocol map[string]bool
+	timeToQuit      time.Duration
 	inpUrlStrChan   chan string
 	masterQuitCh    chan int
 }
@@ -32,6 +34,10 @@ type octopus struct {
 // will pump output onto the implementation's channel returned by its Consume method.
 // CrawlRate is the rate at which requests will be made.
 // RespectRobots (unimplemented) choose whether to respect robots.txt or not.
+// ValidProtocols - This is an array containing the list of url protocols that
+// should be crawled.
+// TimeToQuit - represents the total time to wait between two new nodes to be
+// generated before the crawler quits. This is in seconds.
 type CrawlOptions struct {
 	MaxCrawlDepth      int64
 	MaxCrawlLinks      int64
@@ -41,6 +47,7 @@ type CrawlOptions struct {
 	IncludeBody        bool
 	OpAdapter          OutputAdapter
 	ValidProtocols     []string
+	TimeToQuit         int64
 }
 
 // NodeInfo is used to represent each crawled link and its associated crawl depth.

octopus/pipe_spl_ingest.go

@@ -1,32 +1,50 @@
 package octopus
 
+import (
+	"fmt"
+	"time"
+)
+
 func (o *octopus) makeIngestPipe(inChSet *ingestPipeChSet, opChSet *NodeChSet) {
-	go channelConnector(inChSet, opChSet)
-	go setupStringIngestPipe(inChSet, opChSet)
+	go channelConnector(inChSet, opChSet, o.timeToQuit, o.masterQuitCh)
+	go setupStringIngestPipe(inChSet, opChSet, o.masterQuitCh)
 }
 
-func setupStringIngestPipe(inChSet *ingestPipeChSet, nodeOpChSet *NodeChSet) {
+func setupStringIngestPipe(inChSet *ingestPipeChSet, nodeOpChSet *NodeChSet,
+	masterQuitCh chan int) {
 	for {
 		select {
 		case str := <-inChSet.StrCh:
 			{
 				nodeOpChSet.NodeCh <- createNode("", str, 1)
 			}
-		case i := <-inChSet.QuitCh:
-			{
-				nodeOpChSet.QuitCh <- i
-			}
+		// case i := <-inChSet.QuitCh:
+			// 	{
+			// 		nodeOpChSet.QuitCh <- i
+			// 		masterQuitCh <- i
+			// 	}
 		}
 	}
 }
 
-func channelConnector(inChSet *ingestPipeChSet, opChSet *NodeChSet) {
+func channelConnector(inChSet *ingestPipeChSet, opChSet *NodeChSet,
+	timeOut time.Duration, masterQuitCh chan int) {
 	for {
 		select {
 		case node := <-inChSet.NodeCh:
 			opChSet.NodeCh <- node
 		case i := <-inChSet.QuitCh:
-			opChSet.QuitCh <- i
+			{
+				fmt.Println("Quit Received on Ingest Channel")
+				opChSet.QuitCh <- i
+				masterQuitCh <- i
+			}
+		case <-time.After(timeOut * time.Second):
+			{
+				fmt.Println("Timeout Triggered in Ingest Channel")
+				opChSet.QuitCh <- 1
+				return
+			}
 		}
 	}
 }

octopus/stdpipefunc.go

@@ -1,5 +1,9 @@
 package octopus
 
+import (
+	"fmt"
+)
+
 type stdFunc func(*Node, *NodeChSet)
 
 func stdLinearNodeFunc(stdFn stdFunc, outChSet *NodeChSet) *NodeChSet {
@@ -22,6 +26,7 @@ func stdLinearNodeFunc(stdFn stdFunc, outChSet *NodeChSet) *NodeChSet {
 				}
 			case <-listenQuitCh:
 				{
+					fmt.Println("Quit Received on Internal Channel")
 					outChSet.QuitCh <- 1
 					return
 				}