refactor push event processor
@@ -45,59 +45,75 @@ func NewPushEventProcessor(
}
}
-func (h *pushEventProcessor) Process(ctx context.Context, logger lager.Logger, message *pubsub.Message) (bool, error) {
+func (proc *pushEventProcessor) Process(ctx context.Context, logger lager.Logger, message *pubsub.Message) (bool, error) {
logger = logger.Session("processing-push-event")
- span := h.traceClient.NewSpan("io.pivotal.red.revok/CodePush")
+ span := proc.traceClient.NewSpan("io.pivotal.red.revok/CodePush")
defer span.Finish()
ctx = trace.NewContext(ctx, span)
+ if err := proc.checkSignature(ctx, logger, message); err != nil {
+ return false, err
+ }
+
+ p, err := proc.decodeMessage(ctx, logger, message)
+ if err != nil {
+ return false, err
+ }
+
+ logger = logger.WithData(lager.Data{
+ "owner": p.Owner,
+ "repository": p.Repository,
+ })
+ span.SetLabel("owner", p.Owner)
+ span.SetLabel("repository", p.Repository)
+
+ err = proc.changeFetcher.Fetch(ctx, logger, p.Owner, p.Repository, true)
+ if err != nil {
+ return true, err
+ }
+
+ duration := proc.clock.Since(p.PushTime).Seconds()
+ proc.endToEndGauge.Update(logger, float32(duration))
+
+ return false, nil
+}
+
+func (proc *pushEventProcessor) checkSignature(ctx context.Context, logger lager.Logger, message *pubsub.Message) error {
decodedSignature, err := base64.StdEncoding.DecodeString(message.Attributes["signature"])
if err != nil {
logger.Error("signature-malformed", err, lager.Data{
"signature": message.Attributes["signature"],
})
- return false, err
+ return err
}
- err = h.verifier.Verify(message.Data, decodedSignature)
+ err = proc.verifier.Verify(message.Data, decodedSignature)
if err != nil {
logger.Error("signature-invalid", err, lager.Data{
"signature": message.Attributes["signature"],
})
- h.verifyFailedCounter.Inc(logger)
- return false, err
+ proc.verifyFailedCounter.Inc(logger)
+ return err
}
- decoder := json.NewDecoder(bytes.NewBuffer(message.Data))
+ return nil
+}
+
+func (proc *pushEventProcessor) decodeMessage(ctx context.Context, logger lager.Logger, message *pubsub.Message) (PushEventPlan, error) {
+ buffer := bytes.NewBuffer(message.Data)
var p PushEventPlan
- err = decoder.Decode(&p)
- if err != nil {
+ if err := json.NewDecoder(buffer).Decode(&p); err != nil {
logger.Error("payload-malformed", err)
- return false, err
+ return p, err
}
if p.Owner == "" || p.Repository == "" {
err := errors.New("invalid payload: missing owner or repository")
logger.Error("payload-incomplete", err)
- return false, err
+ return p, err
}
- logger = logger.WithData(lager.Data{
- "owner": p.Owner,
- "repository": p.Repository,
- })
- span.SetLabel("owner", p.Owner)
- span.SetLabel("repository", p.Repository)
-
- err = h.changeFetcher.Fetch(ctx, logger, p.Owner, p.Repository, true)
- if err != nil {
- return true, err
- }
-
- duration := h.clock.Since(p.PushTime).Seconds()
- h.endToEndGauge.Update(logger, float32(duration))
-
- return false, nil
+ return p, nil
}
