Add support to get paginated response

Change the search api, so it returns channels, where we push new message
on each page

Add the -maxPages argument to the command, to limit how many pages
should be paginated

Author: Ajnasz

loggly.go

@@ -29,6 +29,7 @@ const usage = `
     -to <time>        ending time [now]
     -count            print total event count
     -all              print the entire loggly event instead of just the message
+    -maxPages <count> maximum number of pages to query [3]
     -version          print version information
 
   Operators:
@@ -61,6 +62,7 @@ var flags = flag.NewFlagSet("loggly", flag.ExitOnError)
 var count = flags.Bool("count", false, "")
 var versionQuery = flags.Bool("version", false, "")
 var account = flags.String("account", "", "")
+var maxPages = flags.Int("maxPages", 3, "")
 var token = flags.String("token", "", "")
 var size = flags.Int("size", 100, "")
 var from = flags.String("from", "-24h", "")
@@ -89,14 +91,15 @@ func check(err error) {
 }
 
 func printJSON(events []interface{}) error {
-	data, err := json.Marshal(events)
+	for _, event := range events {
+		data, err := json.Marshal(event)
+		if err != nil {
+			return err
+		}
 
-	if err != nil {
-		return err
+		fmt.Println(string(data))
 	}
 
-	fmt.Println(string(data))
-
 	return nil
 }
 
@@ -116,6 +119,53 @@ func printLogMSG(events []interface{}) error {
 	return printJSON(ret)
 }
 
+func execCount(query string, from string, to string) {
+	c := search.New(*account, *token)
+	res, err := c.Query(query).Size(1).From(from).To(to).Fetch()
+	for {
+		select {
+		case r := <-res:
+			fmt.Println(r.Total)
+			return
+		case e := <-err:
+			check(e)
+			return
+		}
+	}
+}
+
+func printRes(res search.Response) {
+	if *allMsg {
+		check(printJSON(res.Events))
+	} else {
+		if err := printLogMSG(res.Events); err != nil {
+			fmt.Fprintf(os.Stderr, "Invalid JSON in the 'logmsg' field. Consider to filter the messages, or use the -all flag and parse the message yourself.\n\n%s", err.Error())
+		}
+	}
+}
+
+func sendQuery(query string, size int, from string, to string, maxPages int) {
+	doneChan := make(chan error)
+
+	c := search.New(*account, *token)
+	res, err := c.Query(query).Size(size).From(from).To(to).MaxPage(maxPages).Fetch()
+
+	go func() {
+		if e := <-err; e != nil {
+			doneChan <- e
+		}
+	}()
+
+	go func() {
+		for i := range res {
+			printRes(i)
+		}
+		doneChan <- nil
+	}()
+
+	check(<-doneChan)
+}
+
 func main() {
 	flags.Usage = printUsage
 	flags.Parse(os.Args[1:])
@@ -130,23 +180,11 @@ func main() {
 
 	args := flags.Args()
 	query := strings.Join(args, " ")
-	c := search.New(*account, *token)
 
 	if *count {
-		res, err := c.Query(query).Size(1).From(*from).To(*to).Fetch()
-		check(err)
-		fmt.Println(res.Total)
-		os.Exit(0)
+		execCount(query, *from, *to)
+		return
 	}
 
-	res, err := c.Query(query).Size(*size).From(*from).To(*to).Fetch()
-	check(err)
-
-	if *allMsg {
-		check(printJSON(res.Events))
-	} else {
-		if err := printLogMSG(res.Events); err != nil {
-			fmt.Fprintf(os.Stderr, "Invalid JSON in the 'logmsg' field. Consider to filter the messages, or use the -all flag and parse the message yourself.\n\n%s", err.Error())
-		}
-	}
+	sendQuery(query, *size, *from, *to, *maxPages)
 }

search/search.go

@@ -28,12 +28,13 @@ type Response struct {
 
 // Query builder struct
 type Query struct {
-	client *Client
-	query  string
-	from   string
-	until  string
-	order  string
-	size   int
+	client   *Client
+	query    string
+	from     string
+	until    string
+	order    string
+	size     int
+	maxPages int
 }
 
 // Create a new query
@@ -59,14 +60,14 @@ func New(account string, token string) *Client {
 	return c
 }
 
-// Url Return the base api url.
-func (c *Client) Url() string {
+// URL Return the base api url.
+func (c *Client) URL() string {
 	return fmt.Sprintf("https://%s.%s", c.Account, c.Endpoint)
 }
 
 // Get the given path.
 func (c *Client) Get(path string) (*http.Response, error) {
-	r, err := http.NewRequest(http.MethodGet, c.Url()+path, nil)
+	r, err := http.NewRequest(http.MethodGet, c.URL()+path, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -76,7 +77,7 @@ func (c *Client) Get(path string) (*http.Response, error) {
 	return client.Do(r)
 }
 
-// Get json from the given path.
+// GetJSON from the given path.
 func (c *Client) GetJSON(path string) (j *simplejson.Json, err error) {
 	res, err := c.Get(path)
 
@@ -114,15 +115,10 @@ func (c *Client) GetEvents(params string) (*simplejson.Json, error) {
 
 // Search response with total events, page number
 // and the events array.
-func (c *Client) Search(params string) (*Response, error) {
-	j, err := c.CreateSearch(params)
-
-	if err != nil {
-		return nil, err
-	}
-
+func (c *Client) Search(j *simplejson.Json, page int) (*Response, error) {
 	id := j.GetPath("rsid", "id").MustString()
-	j, err = c.GetEvents("rsid=" + id)
+
+	j, err := c.GetEvents("rsid=" + id + "&page=" + strconv.Itoa(page))
 
 	if err != nil {
 		return nil, err
@@ -135,6 +131,7 @@ func (c *Client) Search(params string) (*Response, error) {
 		Page:   j.Get("page").MustInt64(),
 		Events: j.Get("events").MustArray(),
 	}, nil
+
 }
 
 // Query Create a new search query using the fluent api.
@@ -165,6 +162,12 @@ func (q *Query) From(str string) *Query {
 	return q
 }
 
+// MaxPage sets the max page
+func (q *Query) MaxPage(maxPages int) *Query {
+	q.maxPages = maxPages
+	return q
+}
+
 // Until Set until time.
 func (q *Query) Until(str string) *Query {
 	q.until = str
@@ -179,6 +182,43 @@ func (q *Query) To(str string) *Query {
 
 // Fetch Search response with total events, page number
 // and the events array.
-func (q *Query) Fetch() (*Response, error) {
-	return q.client.Search(q.String())
+func (q *Query) Fetch() (chan Response, chan error) {
+	resChan := make(chan Response)
+	errChan := make(chan error)
+
+	page := 0
+	go func() {
+		defer close(resChan)
+		defer close(errChan)
+		j, err := q.client.CreateSearch(q.String())
+
+		if err != nil {
+			errChan <- err
+			return
+		}
+
+		for {
+			res, err := q.client.Search(j, page)
+
+			if err != nil {
+				errChan <- err
+				return
+			}
+
+			resChan <- *res
+
+			if page+1 == q.maxPages {
+				return
+			}
+
+			if len(res.Events) < q.size {
+				return
+			}
+
+			page++
+		}
+
+	}()
+
+	return resChan, errChan
 }