Crawl Rate Limiting Feature.

1. Added the Crawl Rate Limiting Feature with capacity for Bursts.
2. Added (currently non-customizable) timeouts for individual requests.
3. Added more documentation to the README and other documentation changes.
4. Added FunctionTags, to identify the specific channel where the quit occurs.
5. General Refactoring.
6. Added error validation to FileOutputAdapter.

Author: droidgod

adapter/basicadapters.go

@@ -73,7 +73,11 @@ func (fw *FileWriterAdapter) writeToFile(listenCh chan *oct.Node,
 		for {
 			select {
 			case output := <-listenCh:
-				fmt.Fprintf(fp, "%d - %s\n", output.Depth, output.UrlString)
+				_, err = fmt.Fprintf(fp, "%d - %s\n", output.Depth,
+					output.UrlString)
+				if err != nil {
+					log.Println("File Error - ", err)
+				}
 			case <-quitCh:
 				return
 			}

octopus/core.go

@@ -2,37 +2,9 @@ package octopus
 
 import (
 	"fmt"
-	"log"
 	"time"
 )
 
-func (o *octopus) setupOctopus() {
-	o.setupValidProtocolMap()
-	o.setupTimeToQuit()
-	o.setupMaxLinksCrawled()
-}
-
-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) * time.Second
-	} else {
-		log.Fatalln("TimeToQuit is not greater than 0")
-	}
-}
-
-func (o *octopus) setupMaxLinksCrawled() {
-	if o.MaxCrawledUrls == 0 {
-		panic("MaxCrawledUrls should either be negative or greater than 0.")
-	}
-}
-
 func (o *octopus) SetupSystem() {
 	o.isReady = false
 	o.setupOctopus()
@@ -57,16 +29,12 @@ func (o *octopus) SetupSystem() {
 	depthLimitChSet := o.makeCrawlDepthFilterPipe(pageParseChSet)
 	maxDelayChSet := o.makeMaxDelayPipe(depthLimitChSet)
 
-	var distributorChSet *NodeChSet
-	if o.MaxCrawledUrls < 0 {
-		distributorChSet = o.makeDistributorPipe(maxDelayChSet, outAdapterChSet)
-	} else {
-		maxLinksCrawledChSet := o.makeLimitCrawlPipe(outAdapterChSet)
-		distributorChSet = o.makeDistributorPipe(maxDelayChSet, maxLinksCrawledChSet)
-	}
+	distributorChSet := o.handleDistributorPipeline(maxDelayChSet, outAdapterChSet)
 
 	pageReqChSet := o.makePageRequisitionPipe(distributorChSet)
-	invUrlFilterChSet := o.makeInvalidUrlFilterPipe(pageReqChSet)
+
+	invUrlFilterChSet := o.handleRateLimitingPipeline(pageReqChSet)
+
 	dupFilterChSet := o.makeDuplicateUrlFilterPipe(invUrlFilterChSet)
 	protoFilterChSet := o.makeUrlProtocolFilterPipe(dupFilterChSet)
 	linkAbsChSet := o.makeLinkAbsolutionPipe(protoFilterChSet)
@@ -77,6 +45,28 @@ func (o *octopus) SetupSystem() {
 	o.isReady = true
 }
 
+func (o *octopus) handleDistributorPipeline(maxDelayChSet, outAdapterChSet *NodeChSet) *NodeChSet {
+	var distributorChSet *NodeChSet
+	if o.MaxCrawledUrls < 0 {
+		distributorChSet = o.makeDistributorPipe(maxDelayChSet, outAdapterChSet)
+	} else {
+		maxLinksCrawledChSet := o.makeCrawlLinkCountLimitPipe(outAdapterChSet)
+		distributorChSet = o.makeDistributorPipe(maxDelayChSet, maxLinksCrawledChSet)
+	}
+	return distributorChSet
+}
+
+func (o *octopus) handleRateLimitingPipeline(pageReqChSet *NodeChSet) *NodeChSet {
+	var invUrlFilterChSet *NodeChSet
+	if o.rateLimiter != nil {
+		rateLimitingChSet := o.makeRateLimitingPipe(pageReqChSet)
+		invUrlFilterChSet = o.makeInvalidUrlFilterPipe(rateLimitingChSet)
+	} else {
+		invUrlFilterChSet = o.makeInvalidUrlFilterPipe(pageReqChSet)
+	}
+	return invUrlFilterChSet
+}
+
 func (o *octopus) BeginCrawling(baseUrlStr string) {
 	if !o.isReady {
 		panic("Call BuildSystem first to setup Octopus")

octopus/doc.go

@@ -24,14 +24,15 @@ The overview of the Pipeline is given below:
 	3. Protocol Filter
 	4. Duplicate Filter
 	5. Invalid Url Filter (Urls whose HEAD request Fails)
-	6. Make GET Request
+	(5x) (Optional) Crawl Rate Limiter.
+	[6]. Make GET Request
 	7a. Send to Output Adapter
 	7b. Check for Timeout (gap between two output on this channel).
 	8. Max Links Crawled Limit Filter
 	9. Depth Limit Filter
 	10. Parse Page for more URLs.
 
 Note: The output from 7b. is fed to 8.
-	1 -> 2 -> 3 -> 4 -> 5 -> 6 -> 7b -> 8 -> 9 -> 10 -> 1
- */
+	1 -> 2 -> 3 -> 4 -> 5 -> (5x) -> [6] -> 7b -> 8 -> 9 -> 10 -> 1
+*/
 package octopus

octopus/modelfactory.go

@@ -3,11 +3,13 @@ package octopus
 import "sync"
 
 const (
-	defaultMaxDepth   int64 = 2
-	anchorTag               = "a"
-	anchorAttrb             = "href"
-	defaultTimeToQuit       = 5
-	defaultCrawlLimit int64 = -1
+	defaultMaxDepth       int64  = 2
+	anchorTag                    = "a"
+	anchorAttrb                  = "href"
+	defaultTimeToQuit            = 30
+	defaultLinkCrawlLimit int64  = -1
+	defaultCrawlRateLimit int64  = -1
+	defaultRequestTimeout uint64 = 15
 )
 
 // NewWithDefaultOptions - Create an Instance of the Octopus with the default CrawlOptions.
@@ -44,15 +46,16 @@ func createNode(parentUrlStr, urlStr string, depth int64) *Node {
 // Returns an instance of CrawlOptions with the values set to sensible defaults.
 func GetDefaultCrawlOptions() *CrawlOptions {
 	return &CrawlOptions{
-		MaxCrawlDepth:      defaultMaxDepth,
-		MaxCrawledUrls:     defaultCrawlLimit,
-		StayWithinBaseHost: false,
-		CrawlRate:          -1,
-		RespectRobots:      false,
-		IncludeBody:        true,
-		OpAdapter:          nil,
-		ValidProtocols:     []string{"http", "https"},
-		TimeToQuit:         defaultTimeToQuit,
+		MaxCrawlDepth:         defaultMaxDepth,
+		MaxCrawledUrls:        defaultLinkCrawlLimit,
+		StayWithinBaseHost:    false,
+		CrawlRatePerSec:       defaultCrawlRateLimit,
+		CrawlBurstLimitPerSec: defaultCrawlRateLimit,
+		RespectRobots:         false,
+		IncludeBody:           true,
+		OpAdapter:             nil,
+		ValidProtocols:        []string{"http", "https"},
+		TimeToQuit:            defaultTimeToQuit,
 	}
 }
 
@@ -65,3 +68,10 @@ func MakeNodeChSet(nodeCh chan<- *Node, quitCh chan<- int) *NodeChSet {
 		},
 	}
 }
+
+// Utility to create a NodeChSet and get full access to the Quit & Node Channel.
+func MakeDefaultNodeChSet() (*NodeChSet, chan *Node, chan int) {
+	nodeCh := make(chan *Node)
+	quitCh := make(chan int)
+	return MakeNodeChSet(nodeCh, quitCh), nodeCh, quitCh
+}

octopus/models.go

@@ -4,6 +4,8 @@ import (
 	"io"
 	"sync"
 	"time"
+
+	"golang.org/x/time/rate"
 )
 
 // octopus is a concurrent web crawler.
@@ -20,6 +22,8 @@ type octopus struct {
 	inputUrlStrChan   chan string
 	masterQuitCh      chan int
 	crawledUrlCounter int64
+	rateLimiter       *rate.Limiter
+	requestTimeout    uint64
 }
 
 // CrawlOptions is used to house options for crawling.
@@ -37,8 +41,11 @@ type octopus struct {
 // 	StayWithinBaseHost - (unimplemented) Ensures crawler stays within the
 // 	level 1 link's hostname.
 //
-// 	CrawlRate (unimplemented) is the rate at which requests will be made.
-// 	In seconds
+// 	CrawlRatePerSec - is the rate at which requests will be made (per second).
+// 	If this is negative, Crawl feature will be ignored. Default is negative.
+//
+// 	CrawlBurstLimitPerSec - Represents the max burst capacity with which requests
+// 	can be made. This must be greater than or equal to the CrawlRatePerSec.
 //
 // 	RespectRobots (unimplemented) choose whether to respect robots.txt or not.
 //
@@ -54,15 +61,16 @@ type octopus struct {
 // 	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
-	MaxCrawledUrls     int64
-	StayWithinBaseHost bool
-	CrawlRate          int64
-	RespectRobots      bool
-	IncludeBody        bool
-	OpAdapter          OutputAdapter
-	ValidProtocols     []string
-	TimeToQuit         int64
+	MaxCrawlDepth         int64
+	MaxCrawledUrls        int64
+	StayWithinBaseHost    bool
+	CrawlRatePerSec       int64
+	CrawlBurstLimitPerSec int64
+	RespectRobots         bool
+	IncludeBody           bool
+	OpAdapter             OutputAdapter
+	ValidProtocols        []string
+	TimeToQuit            int64
 }
 
 // NodeInfo is used to represent each crawled link and its associated crawl depth.

octopus/pipe_augment_linkabsolution.go

@@ -6,7 +6,7 @@ import (
 )
 
 func (o *octopus) makeLinkAbsolutionPipe(outChSet *NodeChSet) *NodeChSet {
-	return stdLinearNodeFunc(makeLinkAbsolute, outChSet)
+	return stdLinearNodeFunc(makeLinkAbsolute, outChSet, "Link Absolution")
 }
 
 func makeLinkAbsolute(node *Node, outChSet *NodeChSet) {

octopus/pipe_ctrl_limitcrawl.go

@@ -4,8 +4,8 @@ import (
 	"sync/atomic"
 )
 
-func (o *octopus) makeLimitCrawlPipe(inChSet *NodeChSet) *NodeChSet {
-	return stdLinearNodeFunc(o.checkWithinLimit, inChSet)
+func (o *octopus) makeCrawlLinkCountLimitPipe(inChSet *NodeChSet) *NodeChSet {
+	return stdLinearNodeFunc(o.checkWithinLimit, inChSet, "Crawl Link Limit")
 }
 
 func (o *octopus) checkWithinLimit(node *Node, outChSet *NodeChSet) {

octopus/pipe_ctrl_ratelimit.go

@@ -0,0 +1,21 @@
+package octopus
+
+import (
+	"time"
+)
+
+func (o *octopus) makeRateLimitingPipe(outChSet *NodeChSet) *NodeChSet {
+	return stdLinearNodeFunc(o.rateLimit, outChSet, "Crawl Rate Limit")
+}
+
+func (o *octopus) rateLimit(node *Node, outChSet *NodeChSet) {
+	for {
+		if r := o.rateLimiter.Reserve(); !r.OK() {
+			continue
+		} else {
+			time.Sleep(r.Delay())
+			outChSet.NodeCh <- node
+			return
+		}
+	}
+}

octopus/pipe_filter_crawldepth.go

@@ -1,7 +1,7 @@
 package octopus
 
 func (o *octopus) makeCrawlDepthFilterPipe(outChSet *NodeChSet) *NodeChSet {
-	return stdLinearNodeFunc(o.filterByUrlDepth, outChSet)
+	return stdLinearNodeFunc(o.filterByUrlDepth, outChSet, "Crawl Depth Filter")
 }
 
 func (o *octopus) filterByUrlDepth(node *Node, outChSet *NodeChSet) {

octopus/pipe_filter_duplicates.go

@@ -1,7 +1,7 @@
 package octopus
 
 func (o *octopus) makeDuplicateUrlFilterPipe(outChSet *NodeChSet) *NodeChSet {
-	return stdLinearNodeFunc(o.filterDuplicates, outChSet)
+	return stdLinearNodeFunc(o.filterDuplicates, outChSet, "Dupliacate Filter")
 }
 
 func (o *octopus) filterDuplicates(node *Node, outChSet *NodeChSet) {