Import source: client

Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.

client.go

@@ -0,0 +1,86 @@
+package reactor_crw
+
+import (
+	"reactor-crw/handler"
+	"strings"
+	"sync"
+
+	"reactor-crw/parser"
+)
+
+// Crawler is an interface for crawler used by the client. It fetches the content
+// sources by the provided path and search parameters which are just a list of
+// required content types. The crawler itself doesn't do anything with fetched
+// data but only collects content sources.
+type Crawler interface {
+	Fetch(path string, search []string) (parser.QueryResult, error)
+}
+
+// Client defines a facade for a specific crawler implementation and a list of
+// content handlers. The client runs the whole process of crawling the data and
+// apply it against a list of provided handlers.
+type Client struct {
+	TotalSources chan int
+	Progress     chan int
+	Errors       chan error
+
+	crawler    Crawler
+	handler    handler.ContentHandler
+	maxWorkers int
+}
+
+// NewClient creates a new crawler client and initialize client channels.
+func NewClient(c Crawler, maxWorkers int, ch handler.ContentHandler) *Client {
+	return &Client{
+		crawler:      c,
+		handler:      ch,
+		maxWorkers:   maxWorkers,
+		TotalSources: make(chan int, 1),
+		Progress:     make(chan int),
+		Errors:       make(chan error),
+	}
+}
+
+// Run starts all the work by running Crawler and processes its search results
+// with the corresponding handler.ContentHandler.
+//
+// All progress and errors signals will be sent to Progress and Errors channels
+// respectively. TotalSources channel is used to notify the client about the
+// amount of found content links.
+//
+// Content sources will be processed by handler.ContentHandler through a simple
+// worker pool.
+func (c *Client) Run(path string, search string) error {
+	defer func() {
+		close(c.Progress)
+		close(c.Errors)
+	}()
+
+	collectedData, err := c.crawler.Fetch(path, strings.Split(search, ","))
+	c.TotalSources <- len(collectedData)
+	if err != nil || len(collectedData) == 0 {
+		return err
+	}
+
+	contentHandlerTasks := make(chan string, len(collectedData))
+	for task := range collectedData {
+		contentHandlerTasks <- task
+	}
+	close(contentHandlerTasks)
+
+	var wg sync.WaitGroup
+
+	for i := 0; i < c.maxWorkers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for t := range contentHandlerTasks {
+				c.handler.Process(t, c.Progress, c.Errors)
+			}
+		}()
+	}
+
+	wg.Wait()
+
+	return nil
+}