Matching Service should not serve requests before it's ready
Issue
package matching
import (
+ "sync"
+
m "github.com/uber/cadence/.gen/go/matching"
gen "github.com/uber/cadence/.gen/go/shared"
"github.com/uber/cadence/common/persistence"
@@ -14,6 +16,7 @@ var _ m.TChanMatchingService = (*Handler)(nil)
type Handler struct {
taskPersistence persistence.TaskManager
engine Engine
+ startWG sync.WaitGroup
service.Service
}
@@ -23,6 +26,8 @@ func NewHandler(taskPersistence persistence.TaskManager, sVice service.Service)
Service: sVice,
taskPersistence: taskPersistence,
}
+ // prevent us from trying to serve requests before matching engine is started and ready
+ handler.startWG.Add(1)
return handler, []thrift.TChanServer{m.NewTChanMatchingServiceServer(handler)}
}
@@ -34,6 +39,7 @@ func (h *Handler) Start(thriftService []thrift.TChanServer) error {
return err
}
h.engine = NewEngine(h.taskPersistence, history, h.Service.GetLogger())
+ h.startWG.Done()
return nil
}
@@ -51,12 +57,14 @@ func (h *Handler) IsHealthy(ctx thrift.Context) (bool, error) {
// AddActivityTask - adds an activity task.
func (h *Handler) AddActivityTask(ctx thrift.Context, addRequest *m.AddActivityTaskRequest) error {
h.Service.GetLogger().Debug("Engine Received AddActivityTask")
+ h.startWG.Wait()
return h.engine.AddActivityTask(addRequest)
}
// AddDecisionTask - adds a decision task.
func (h *Handler) AddDecisionTask(ctx thrift.Context, addRequest *m.AddDecisionTaskRequest) error {
h.Service.GetLogger().Debug("Engine Received AddDecisionTask")
+ h.startWG.Wait()
return h.engine.AddDecisionTask(addRequest)
}
@@ -64,6 +72,7 @@ func (h *Handler) AddDecisionTask(ctx thrift.Context, addRequest *m.AddDecisionT
func (h *Handler) PollForActivityTask(ctx thrift.Context,
pollRequest *gen.PollForActivityTaskRequest) (*gen.PollForActivityTaskResponse, error) {
h.Service.GetLogger().Debug("Engine Received PollForActivityTask")
+ h.startWG.Wait()
response, error := h.engine.PollForActivityTask(ctx, pollRequest)
h.Service.GetLogger().Debug("Engine returned from PollForActivityTask")
return response, error
@@ -74,6 +83,7 @@ func (h *Handler) PollForActivityTask(ctx thrift.Context,
func (h *Handler) PollForDecisionTask(ctx thrift.Context,
pollRequest *gen.PollForDecisionTaskRequest) (*gen.PollForDecisionTaskResponse, error) {
h.Service.GetLogger().Debug("Engine Received PollForDecisionTask")
+ h.startWG.Wait()
response, error := h.engine.PollForDecisionTask(ctx, pollRequest)
h.Service.GetLogger().Debug("Engine returned from PollForDecisionTask")
return response, error
