diff --git a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitor.java b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitor.java
index 1dc3df348..5d6129939 100644
--- a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitor.java
+++ b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/BuildEventMonitor.java
@@ -19,7 +19,6 @@
 
 import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.anyArtifactsMatchExpected;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.spectator.api.Registry;
 import com.netflix.spinnaker.echo.model.Event;
 import com.netflix.spinnaker.echo.model.Pipeline;
@@ -38,7 +37,6 @@
 import lombok.val;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-import rx.Observable;
 import rx.functions.Action1;
 
 /**
@@ -49,30 +47,21 @@ public class BuildEventMonitor extends TriggerMonitor {
 
   public static final String[] BUILD_TRIGGER_TYPES = {"jenkins", "travis", "wercker"};
 
-  private final ObjectMapper objectMapper = new ObjectMapper();
-
-  private final PipelineCache pipelineCache;
-
   @Autowired
   public BuildEventMonitor(@NonNull PipelineCache pipelineCache,
                            @NonNull Action1<Pipeline> subscriber,
                            @NonNull Registry registry) {
-    super(subscriber, registry);
-    this.pipelineCache = pipelineCache;
+    super(pipelineCache, subscriber, registry);
   }
 
   @Override
-  public void processEvent(Event event) {
-    super.validateEvent(event);
-    if (!event.getDetails().getType().equalsIgnoreCase(BuildEvent.TYPE)) {
-      return;
-    }
+  protected boolean handleEventType(String eventType) {
+    return eventType.equalsIgnoreCase(BuildEvent.TYPE);
+  }
 
-    BuildEvent buildEvent = objectMapper.convertValue(event, BuildEvent.class);
-    Observable.just(buildEvent)
-      .doOnNext(this::onEchoResponse)
-      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
-      .subscribe(triggerEachMatch());
+  @Override
+  protected BuildEvent convertEvent(Event event) {
+    return objectMapper.convertValue(event, BuildEvent.class);
   }
 
   @Override
diff --git a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitor.java b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitor.java
index 6dbc16960..487995f3d 100644
--- a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitor.java
+++ b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/DockerEventMonitor.java
@@ -16,7 +16,8 @@
 
 package com.netflix.spinnaker.echo.pipelinetriggers.monitor;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
+import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.anyArtifactsMatchExpected;
+
 import com.netflix.spectator.api.Registry;
 import com.netflix.spinnaker.echo.model.Event;
 import com.netflix.spinnaker.echo.model.Pipeline;
@@ -25,51 +26,38 @@
 import com.netflix.spinnaker.echo.model.trigger.TriggerEvent;
 import com.netflix.spinnaker.echo.pipelinetriggers.PipelineCache;
 import com.netflix.spinnaker.kork.artifacts.model.Artifact;
+import java.util.Collections;
+import java.util.List;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.regex.PatternSyntaxException;
 import lombok.NonNull;
 import lombok.val;
 import org.apache.commons.lang.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-import rx.Observable;
 import rx.functions.Action1;
 
-import java.util.Collections;
-import java.util.List;
-import java.util.function.Function;
-import java.util.function.Predicate;
-import java.util.regex.PatternSyntaxException;
-
-import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.anyArtifactsMatchExpected;
-
 @Component
 public class DockerEventMonitor extends TriggerMonitor {
 
   public static final String TRIGGER_TYPE = "docker";
 
-  private final ObjectMapper objectMapper = new ObjectMapper();
-
-  private final PipelineCache pipelineCache;
-
   @Autowired
   public DockerEventMonitor(@NonNull PipelineCache pipelineCache,
                             @NonNull Action1<Pipeline> subscriber,
                             @NonNull Registry registry) {
-    super(subscriber, registry);
-    this.pipelineCache = pipelineCache;
+    super(pipelineCache, subscriber, registry);
   }
 
   @Override
-  public void processEvent(Event event) {
-    super.validateEvent(event);
-    if (!event.getDetails().getType().equalsIgnoreCase(DockerEvent.TYPE)) {
-      return;
-    }
+  protected boolean handleEventType(String eventType) {
+    return eventType.equalsIgnoreCase(DockerEvent.TYPE);
+  }
 
-    DockerEvent dockerEvent = objectMapper.convertValue(event, DockerEvent.class);
-    Observable.just(dockerEvent)
-      .doOnNext(this::onEchoResponse)
-      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
-      .subscribe(triggerEachMatch());
+  @Override
+  protected DockerEvent convertEvent(Event event) {
+    return objectMapper.convertValue(event, DockerEvent.class);
   }
 
   @Override
diff --git a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/GitEventMonitor.java b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/GitEventMonitor.java
index 030b333f2..5d59047ba 100644
--- a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/GitEventMonitor.java
+++ b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/GitEventMonitor.java
@@ -18,7 +18,6 @@
 
 import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.anyArtifactsMatchExpected;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.spectator.api.Registry;
 import com.netflix.spinnaker.echo.model.Event;
 import com.netflix.spinnaker.echo.model.Pipeline;
@@ -38,7 +37,6 @@
 import org.apache.commons.lang.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-import rx.Observable;
 import rx.functions.Action1;
 
 /**
@@ -52,30 +50,22 @@ public class GitEventMonitor extends TriggerMonitor {
 
   private static final String GITHUB_SECURE_SIGNATURE_HEADER = "X-Hub-Signature";
 
-  private final ObjectMapper objectMapper = new ObjectMapper();
-
-  private final PipelineCache pipelineCache;
-
   @Autowired
   public GitEventMonitor(@NonNull PipelineCache pipelineCache,
                          @NonNull Action1<Pipeline> subscriber,
                          @NonNull Registry registry) {
-    super(subscriber, registry);
-    this.pipelineCache = pipelineCache;
+    super(pipelineCache, subscriber, registry);
   }
 
   @Override
-  public void processEvent(Event event) {
-    super.validateEvent(event);
-    if (!event.getDetails().getType().equalsIgnoreCase(GitEvent.TYPE)) {
-      return;
-    }
+  protected boolean handleEventType(String eventType) {
+    return eventType.equalsIgnoreCase(GitEvent.TYPE);
+  }
+
 
-    GitEvent buildEvent = objectMapper.convertValue(event, GitEvent.class);
-    Observable.just(buildEvent)
-      .doOnNext(this::onEchoResponse)
-      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
-      .subscribe(triggerEachMatch());
+  @Override
+  protected GitEvent convertEvent(Event event) {
+    return objectMapper.convertValue(event, GitEvent.class);
   }
 
   @Override
diff --git a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/PubsubEventMonitor.java b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/PubsubEventMonitor.java
index d37d31f6f..0041986a9 100644
--- a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/PubsubEventMonitor.java
+++ b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/PubsubEventMonitor.java
@@ -19,7 +19,6 @@
 import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.anyArtifactsMatchExpected;
 import static com.netflix.spinnaker.echo.pipelinetriggers.artifacts.ArtifactMatcher.isConstraintInPayload;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.spectator.api.Registry;
 import com.netflix.spinnaker.echo.model.Event;
 import com.netflix.spinnaker.echo.model.Pipeline;
@@ -36,7 +35,6 @@
 import lombok.extern.slf4j.Slf4j;
 import lombok.val;
 import org.apache.commons.lang.StringUtils;
-import rx.Observable;
 import rx.functions.Action1;
 
 /**
@@ -47,30 +45,21 @@ public class PubsubEventMonitor extends TriggerMonitor {
 
   public static final String PUBSUB_TRIGGER_TYPE = "pubsub";
 
-  private final ObjectMapper objectMapper = new ObjectMapper();
-
-  private final PipelineCache pipelineCache;
-
   public PubsubEventMonitor(@NonNull PipelineCache pipelineCache,
                             @NonNull Action1<Pipeline> subscriber,
                             @NonNull Registry registry) {
-    super(subscriber, registry);
-    this.pipelineCache = pipelineCache;
+    super(pipelineCache, subscriber, registry);
   }
 
   @Override
-  public void processEvent(Event event) {
-    super.validateEvent(event);
-    if (!event.getDetails().getType().equalsIgnoreCase(PubsubEvent.TYPE)) {
-      return;
-    }
+  protected boolean handleEventType(String eventType) {
+    return eventType.equalsIgnoreCase(PubsubEvent.TYPE);
+  }
 
-    PubsubEvent pubsubEvent = objectMapper.convertValue(event, PubsubEvent.class);
 
-    Observable.just(pubsubEvent)
-      .doOnNext(this::onEchoResponse)
-      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
-      .subscribe(triggerEachMatch());
+  @Override
+  protected PubsubEvent convertEvent(Event event) {
+    return objectMapper.convertValue(event, PubsubEvent.class);
   }
 
   @Override
diff --git a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/TriggerMonitor.java b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/TriggerMonitor.java
index df74cf9a7..2be2266b6 100644
--- a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/TriggerMonitor.java
+++ b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/TriggerMonitor.java
@@ -16,12 +16,14 @@
 
 package com.netflix.spinnaker.echo.pipelinetriggers.monitor;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.spectator.api.Registry;
 import com.netflix.spinnaker.echo.events.EchoEventListener;
 import com.netflix.spinnaker.echo.model.Event;
 import com.netflix.spinnaker.echo.model.Pipeline;
 import com.netflix.spinnaker.echo.model.Trigger;
 import com.netflix.spinnaker.echo.model.trigger.TriggerEvent;
+import com.netflix.spinnaker.echo.pipelinetriggers.PipelineCache;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.Function;
@@ -50,6 +52,8 @@ static class TriggerMatchParameters {
 
   protected final Action1<Pipeline> subscriber;
   protected final Registry registry;
+  protected final ObjectMapper objectMapper = new ObjectMapper();
+  protected final PipelineCache pipelineCache;
 
   protected void validateEvent(Event event) {
     if (event.getDetails() == null) {
@@ -59,10 +63,25 @@ protected void validateEvent(Event event) {
     }
   }
 
-  public TriggerMonitor(@NonNull Action1<Pipeline> subscriber,
+  public TriggerMonitor(@NonNull PipelineCache pipelineCache,
+                        @NonNull Action1<Pipeline> subscriber,
                         @NonNull Registry registry) {
     this.subscriber = subscriber;
     this.registry = registry;
+    this.pipelineCache = pipelineCache;
+  }
+
+  public void processEvent(Event event) {
+    validateEvent(event);
+    if (!handleEventType(event.getDetails().getType())) {
+      return;
+    }
+
+    TriggerEvent triggerEvent = convertEvent(event);
+    Observable.just(triggerEvent)
+      .doOnNext(this::onEchoResponse)
+      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
+      .subscribe(triggerEachMatch());
   }
 
   protected boolean matchesPattern(String s, String pattern) {
@@ -112,6 +131,10 @@ protected Func1<Pipeline, Optional<Pipeline>> withMatchingTrigger(final TriggerE
     };
   }
 
+  protected abstract boolean handleEventType(String eventType);
+
+  protected abstract TriggerEvent convertEvent(Event event);
+
   protected abstract boolean isSuccessfulTriggerEvent(TriggerEvent event);
 
   protected abstract Predicate<Trigger> matchTriggerFor(final TriggerEvent event, final Pipeline pipeline);
diff --git a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/WebhookEventMonitor.java b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/WebhookEventMonitor.java
index 341b0ac1c..224ba0120 100644
--- a/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/WebhookEventMonitor.java
+++ b/echo-pipelinetriggers/src/main/java/com/netflix/spinnaker/echo/pipelinetriggers/monitor/WebhookEventMonitor.java
@@ -18,11 +18,12 @@
 import static java.util.Collections.emptyList;
 
 import com.fasterxml.jackson.core.type.TypeReference;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.spectator.api.Registry;
 import com.netflix.spinnaker.echo.model.Event;
 import com.netflix.spinnaker.echo.model.Pipeline;
 import com.netflix.spinnaker.echo.model.Trigger;
+import com.netflix.spinnaker.echo.model.trigger.DockerEvent;
+import com.netflix.spinnaker.echo.model.trigger.PubsubEvent;
 import com.netflix.spinnaker.echo.model.trigger.TriggerEvent;
 import com.netflix.spinnaker.echo.model.trigger.WebhookEvent;
 import com.netflix.spinnaker.echo.pipelinetriggers.PipelineCache;
@@ -38,7 +39,6 @@
 import lombok.val;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-import rx.Observable;
 import rx.functions.Action1;
 
 @Component @Slf4j
@@ -49,32 +49,22 @@ public class WebhookEventMonitor extends TriggerMonitor {
   private static final TypeReference<List<Artifact>> ARTIFACT_LIST =
       new TypeReference<List<Artifact>>() {};
 
-  private final ObjectMapper objectMapper = new ObjectMapper();
-
-  private final PipelineCache pipelineCache;
-
   @Autowired
   public WebhookEventMonitor(@NonNull PipelineCache pipelineCache,
                              @NonNull Action1<Pipeline> subscriber,
                              @NonNull Registry registry) {
-    super(subscriber, registry);
-    this.pipelineCache = pipelineCache;
+    super(pipelineCache, subscriber, registry);
   }
 
   @Override
-  public void processEvent(Event event) {
-    super.validateEvent(event);
-    if (event.getDetails().getType() == null) {
-      return;
-    }
+  protected boolean handleEventType(String eventType) {
+    return eventType != null;
+  }
 
-    /* Need to create WebhookEvent, since TriggerEvent is abstract */
-    WebhookEvent webhookEvent = objectMapper.convertValue(event, WebhookEvent.class);
 
-    Observable.just(webhookEvent)
-      .doOnNext(this::onEchoResponse)
-      .zipWith(pipelineCache.getPipelines(), TriggerMatchParameters::new)
-      .subscribe(triggerEachMatch());
+  @Override
+  protected WebhookEvent convertEvent(Event event) {
+    return objectMapper.convertValue(event, WebhookEvent.class);
   }
 
   @Override