add the ability to have the type in the routing key for rabbitmq prod…

src/main/java/com/zendesk/maxwell/producer/MaxwellKafkaProducer.java

@@ -235,11 +235,12 @@ private Integer getNumPartitions(String topic) {
 		}
 	}
 
-	private String generateTopic(String topic, RowIdentity pk){
-		if ( interpolateTopic )
-			return topic.replaceAll("%\\{database\\}", pk.getDatabase()).replaceAll("%\\{table\\}", pk.getTable());
-		else
+	private String generateTopic(String topic, RowMap rowMap){
+		if(rowMap == null || !interpolateTopic){
 			return topic;
+		}
+		return rowMap.buildDestinationString(topic, "%\\{database}",
+				"%\\{table}", "%\\{type}");
 	}
 
 	@Override
@@ -273,7 +274,6 @@ void sendAsync(ProducerRecord<String, String> record, Callback callback) {
 	}
 
 	ProducerRecord<String, String> makeProducerRecord(final RowMap r) throws Exception {
-		RowIdentity pk = r.getRowIdentity();
 		String key = r.pkToJson(keyFormat);
 		String value = r.toJSON(outputConfig);
 		ProducerRecord<String, String> record;
@@ -285,7 +285,7 @@ record = new ProducerRecord<>(this.ddlTopic, this.ddlPartitioner.kafkaPartition(
 			// javascript topic override
 			topic = r.getKafkaTopic();
 			if ( topic == null ) {
-				topic = generateTopic(this.topic, pk);
+				topic = generateTopic(this.topic, r);
 			}
 
 			record = new ProducerRecord<>(topic, this.partitioner.kafkaPartition(r, getNumPartitions(topic)), key, value);
@@ -296,7 +296,7 @@ record = new ProducerRecord<>(topic, this.partitioner.kafkaPartition(r, getNumPa
 	ProducerRecord<String, String> makeFallbackRecord(String fallbackTopic, final RowIdentity pk, Exception reason) throws Exception {
 		String key = pk.toKeyJson(keyFormat);
 		String value = pk.toFallbackValueWithReason(reason.getClass().getSimpleName());
-		String topic = generateTopic(fallbackTopic, pk);
+		String topic = generateTopic(fallbackTopic, queue.peek());
 		return new ProducerRecord<>(topic, key, value);
 	}
 

src/main/java/com/zendesk/maxwell/producer/MaxwellRedisProducer.java

@@ -1,7 +1,6 @@
 package com.zendesk.maxwell.producer;
 
 import com.zendesk.maxwell.MaxwellContext;
-import com.zendesk.maxwell.row.RowIdentity;
 import com.zendesk.maxwell.row.RowMap;
 import com.zendesk.maxwell.util.StoppableTask;
 import org.slf4j.Logger;
@@ -43,18 +42,17 @@ public MaxwellRedisProducer(MaxwellContext context) {
 		}
 	}
 
-	private String generateChannel(RowIdentity pk){
-		if (interpolateChannel) {
-			return channel.replaceAll("%\\{database}", pk.getDatabase()).replaceAll("%\\{table}", pk.getTable());
+	private String generateChannel(RowMap rowMap){
+		if(interpolateChannel) {
+			return rowMap.buildDestinationString(channel, "%\\{database}", "%\\{table}", "%\\{type}");
 		}
-
 		return channel;
 	}
 
 	private void sendToRedis(RowMap msg) throws Exception {
 		String messageStr = msg.toJSON(outputConfig);
 
-		String channel = this.generateChannel(msg.getRowIdentity());
+		String channel = this.generateChannel(msg);
 
 		switch (redisType) {
 			case "lpush":

src/main/java/com/zendesk/maxwell/producer/RabbitmqProducer.java

@@ -6,6 +6,7 @@
 import com.rabbitmq.client.MessageProperties;
 import com.zendesk.maxwell.MaxwellContext;
 import com.zendesk.maxwell.row.RowMap;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -60,15 +61,8 @@ public void push(RowMap r) throws Exception {
 	}
 
 	private String getRoutingKeyFromTemplate(RowMap r) {
-		String table = r.getTable();
-
-		if ( table == null )
-			table = "";
-
-		return context
-				.getConfig()
-				.rabbitmqRoutingKeyTemplate
-				.replace("%db%", r.getDatabase())
-				.replace("%table%", table);
+		String destination = context.getConfig().rabbitmqRoutingKeyTemplate;
+		return r.buildDestinationString(destination,
+				"%db%", "%table%", "%type%");
 	}
 }

src/main/java/com/zendesk/maxwell/row/RowIdentity.java

@@ -28,6 +28,10 @@ public String getTable() {
 		return table;
 	}
 
+	public String getRowType() {
+		return rowType;
+	}
+
 	public String toKeyJson(RowMap.KeyFormat keyFormat) throws IOException {
 		MaxwellJson json = MaxwellJson.getInstance();
 		JsonGenerator g = json.reset();

src/main/java/com/zendesk/maxwell/row/RowMap.java

@@ -5,6 +5,7 @@
 import com.zendesk.maxwell.replication.BinlogPosition;
 import com.zendesk.maxwell.producer.MaxwellOutputConfig;
 import com.zendesk.maxwell.replication.Position;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -425,4 +426,21 @@ public String getComment() {
 	public void setComment(String comment) {
 		this.comment = comment;
 	}
+
+  /**
+	 * Used by Producers to build the destination String (Topic, Channel, RoutingKey)
+	 *
+	 * @param destination The default destination String
+	 * @param databaseRegex database regex to be replaced by the database name
+	 * @param tableRegex The table regex to be replaced by the table name
+	 * @param typeRegex The type regex to be replaced by the type
+	 * @return The destination string with all the regex's replaced by their corresponding types
+	 */
+	public String buildDestinationString(String destination, String databaseRegex, String tableRegex,
+			String typeRegex) {
+		return destination
+				.replaceAll(databaseRegex, StringUtils.defaultString(this.database, StringUtils.EMPTY))
+				.replaceAll(tableRegex, StringUtils.defaultString(this.table, StringUtils.EMPTY))
+				.replaceAll(typeRegex, StringUtils.defaultString(this.rowType, StringUtils.EMPTY));
+  }
 }