### Question:
ProductCategoriesCounter extends AbstractMessageSplitter { @Override protected Object splitMessage(Message<?> message) { List<Tuple> output = new ArrayList<Tuple>(); List<Tuple> input = (List<Tuple>) message.getPayload(); Map<String,Tuple> processed = new HashMap<>(); for (Tuple t : input) { String product = t.getString("product"); String category = t.getString("category"); int count = count(product, category); Tuple previous = processed.get(buildLocalKey(category, product)); if (previous != null) { output.remove(previous); } Tuple tcount = TupleBuilder.tuple().of("count", count, "product", product, "category", category); output.add(tcount); processed.put(buildLocalKey(category,product),tcount); } return output; } ProductCategoriesCounter(RedisTemplate<String, String> redisTemplate); int getProductCategoryCount(String category, String product);  }### Answer:
@Test public void test() { List<Tuple> history = new ArrayList<>(); history.add(TupleBuilder.tuple().of("product", "8", "category", "Players")); history.add(TupleBuilder.tuple().of("product", "8", "category", "Players")); history.add(TupleBuilder.tuple().of("product", "8", "category", "TVs")); history.add(TupleBuilder.tuple().of("product", "8", "category", "Mounts")); history.add(TupleBuilder.tuple().of("product", "0", "category", "Phones")); history.add(TupleBuilder.tuple().of("product", "2", "category", "Phones")); history.add(TupleBuilder.tuple().of("product", "17", "category", "Phones")); history.add(TupleBuilder.tuple().of("product", "21", "category", "Phones")); List<Tuple> results = (List<Tuple>) productCategoriesCounter.splitMessage(new GenericMessage<List<Tuple>> (history)); assertThat(results.size(), is(7)); for (Tuple t : results) { if ("Players".equals(t.getString("category"))) { assertThat(t.getInt("count"), is(2)); } else { assertThat(t.getInt("count"), is(1)); } } Map<Object, Object> entries; entries = redisTemplate.boundHashOps("prodcnt:8").entries(); assertEquals(3, entries.size()); assertEquals("2", entries.get("Players")); assertEquals("1", entries.get("TVs")); assertEquals("1", entries.get("Mounts")); entries = redisTemplate.boundHashOps("prodcnt:0").entries(); assertEquals(1, entries.size()); assertEquals("1", entries.get("Phones")); entries = redisTemplate.boundHashOps("prodcnt:2").entries(); assertEquals(1, entries.size()); assertEquals("1", entries.get("Phones")); entries = redisTemplate.boundHashOps("prodcnt:17").entries(); assertEquals(1, entries.size()); assertEquals("1", entries.get("Phones")); entries = redisTemplate.boundHashOps("prodcnt:21").entries(); assertEquals(1, entries.size()); assertEquals("1", entries.get("Phones")); }