producer republish command

Gopkg.toml

@@ -17,7 +17,7 @@
 
 [[constraint]]
   name = "gopkg.in/src-d/framework.v0"
-  revision = "0839a41303d85dd722c2bf23106f01e0f6758115"
+  revision = "88f5399ee0037158edaef606ec57b7037a80c02a"
   source = "https://github.com/src-d/framework.git"
 
 [[constraint]]

cli/borges/init.go

@@ -3,8 +3,6 @@ package main
 import (
 	"fmt"
 
-	"github.com/inconshreveable/log15"
-
 	"gopkg.in/src-d/core-retrieval.v0/schema"
 	"gopkg.in/src-d/framework.v0/database"
 )
@@ -38,7 +36,7 @@ func (c *initCmd) Execute(args []string) error {
 		return fmt.Errorf("unable to create database schema: %s", err)
 	}
 
-	log15.Info("database was successfully initialized")
+	log.Info("database was successfully initialized")
 	return nil
 }
 

cli/borges/producer.go

@@ -105,6 +105,7 @@ func checkPriority(prio uint8) error {
 var producerSubcommands = []ExecutableCommand{
 	mentionsCommand,
 	fileCommand,
+	republishCommand,
 }
 
 func init() {

cli/borges/republish.go

@@ -0,0 +1,72 @@
+package main
+
+import (
+	"time"
+
+	"github.com/src-d/borges"
+	"gopkg.in/src-d/framework.v0/queue"
+)
+
+const (
+	republishCmdName      = "republish"
+	republishCmdShortDesc = "requeue jobs from buried queues"
+	republishCmdLongDesc  = ""
+)
+
+// republishCommand is a producer subcommand.
+var republishCommand = &republishCmd{producerSubcmd: newProducerSubcmd(
+	republishCmdName,
+	republishCmdShortDesc,
+	republishCmdLongDesc,
+)}
+
+type republishCmd struct {
+	producerSubcmd
+
+	Time string `long:"time" short:"t" default:"0" description:"elapsed time between republish triggers"`
+}
+
+func (c *republishCmd) Execute(args []string) error {
+	lapse, err := time.ParseDuration(c.Time)
+	if err != nil {
+		return err
+	}
+
+	if err := c.producerSubcmd.init(); err != nil {
+		return err
+	}
+	defer c.broker.Close()
+
+	log.Info("starting republishing jobs...", "time", c.Time)
+
+	log.Debug("republish task triggered ")
+	if err := c.queue.RepublishBuried(republishCondition); err != nil {
+		log.Error("error republishing buried jobs", "error", err)
+	}
+
+	if lapse != 0 {
+		c.runPeriodically(lapse)
+	}
+
+	log.Info("stopping republishing jobs")
+	return nil
+}
+
+func republishCondition(job *queue.Job) bool {
+	// Althoug the job has the temporary error tag, it must be checked
+	// that the retries is equals to zero. The reason for this is that
+	// a job can panic during a retry process, so it can be tagged as
+	// temporary error and a number of retries greater than zero reveals
+	// that fact.
+	return job.ErrorType == borges.TemporaryError && job.Retries == 0
+}
+
+func (c *republishCmd) runPeriodically(lapse time.Duration) {
+	ticker := time.Tick(lapse)
+	for range ticker {
+		log.Debug("republish task triggered ")
+		if err := c.queue.RepublishBuried(republishCondition); err != nil {
+			log.Error("error republishing buried jobs", "error", err)
+		}
+	}
+}

cli/borges/update.go

@@ -0,0 +1,23 @@
+package main
+
+const (
+	updateCmdName      = "update"
+	updateCmdShortName = "update repositories processed previously"
+	updateCmdLongDesc  = ""
+)
+
+// updateCommand is a producer subcommand.
+var updateCommand = &updateCmd{producerSubcmd: newProducerSubcmd(
+	updateCmdName,
+	updateCmdShortName,
+	updateCmdLongDesc,
+)}
+
+type updateCmd struct {
+	producerSubcmd
+}
+
+func (c *updateCmd) Execute(args []string) error {
+	log.Warn("Update command is not implemented yet")
+	return nil
+}