Merge pull request #176 from takonomura/notify-job

Notify to starter when new job enqueued

cmd/server/cmd.go

@@ -46,13 +46,15 @@ type myShoes struct {
 
 // newShoes create myshoes.
 func newShoes() (*myShoes, error) {
-	ds, err := mysql.New(config.Config.MySQLDSN)
+	notifyEnqueueCh := make(chan struct{}, 1)
+
+	ds, err := mysql.New(config.Config.MySQLDSN, notifyEnqueueCh)
 	if err != nil {
 		return nil, fmt.Errorf("failed to mysql.New: %w", err)
 	}
 
 	unlimit := unlimited.Unlimited{}
-	s := starter.New(ds, unlimit, config.Config.RunnerVersion)
+	s := starter.New(ds, unlimit, config.Config.RunnerVersion, notifyEnqueueCh)
 
 	manager := runner.New(ds, config.Config.RunnerVersion)
 

internal/testutils/testutils.go

@@ -43,7 +43,7 @@ func IntegrationTestRunner(m *testing.M) int {
 	if err := pool.Retry(func() error {
 		var err error
 		dsn := fmt.Sprintf("root:%s@(localhost:%s)/mysql", mysqlRootPassword, resource.GetPort("3306/tcp"))
-		testDatastore, err = mysql.New(dsn)
+		testDatastore, err = mysql.New(dsn, make(chan<- struct{}))
 		if err != nil {
 			log.Fatalf("failed to create datastore instance: %s", err)
 		}

pkg/datastore/mysql/job.go

@@ -17,6 +17,13 @@ func (m *MySQL) EnqueueJob(ctx context.Context, job datastore.Job) error {
 		return fmt.Errorf("failed to execute INSERT query: %w", err)
 	}
 
+	select {
+	case m.notifyEnqueueCh <- struct{}{}:
+		// notified to starter
+	default:
+		// no capacity on channel, do not block
+	}
+
 	return nil
 }
 

pkg/datastore/mysql/mysql.go

@@ -11,10 +11,12 @@ import (
 // MySQL is implement datastore in MySQL
 type MySQL struct {
 	Conn *sqlx.DB
+
+	notifyEnqueueCh chan<- struct{}
 }
 
 // New create mysql connection
-func New(dsn string) (*MySQL, error) {
+func New(dsn string, notifyEnqueueCh chan<- struct{}) (*MySQL, error) {
 	u, err := getMySQLURL(dsn)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get MySQL URL: %w", err)
@@ -26,7 +28,8 @@ func New(dsn string) (*MySQL, error) {
 	}
 
 	return &MySQL{
-		Conn: conn,
+		Conn:            conn,
+		notifyEnqueueCh: notifyEnqueueCh,
 	}, nil
 }
 

pkg/starter/starter.go

@@ -39,17 +39,19 @@ var (
 
 // Starter is dispatcher for running job
 type Starter struct {
-	ds            datastore.Datastore
-	safety        safety.Safety
-	runnerVersion string
+	ds              datastore.Datastore
+	safety          safety.Safety
+	runnerVersion   string
+	notifyEnqueueCh <-chan struct{}
 }
 
 // New create starter instance
-func New(ds datastore.Datastore, s safety.Safety, runnerVersion string) *Starter {
+func New(ds datastore.Datastore, s safety.Safety, runnerVersion string, notifyEnqueueCh <-chan struct{}) *Starter {
 	return &Starter{
-		ds:            ds,
-		safety:        s,
-		runnerVersion: runnerVersion,
+		ds:              ds,
+		safety:          s,
+		runnerVersion:   runnerVersion,
+		notifyEnqueueCh: notifyEnqueueCh,
 	}
 }
 
@@ -82,6 +84,11 @@ func (s *Starter) Loop(ctx context.Context) error {
 				if err := s.dispatcher(ctx, ch); err != nil {
 					logger.Logf(false, "failed to starter: %+v", err)
 				}
+			case <-s.notifyEnqueueCh:
+				ticker.Reset(10 * time.Second)
+				if err := s.dispatcher(ctx, ch); err != nil {
+					logger.Logf(false, "failed to starter: %+v", err)
+				}
 			case <-ctx.Done():
 				return nil
 			}