CassandraBolt implements Serializable { @SuppressWarnings("unchecked") public void prepare(Map<String, Object> stormConf, TopologyContext context) { Map<String, Object> config = (Map<String, Object>) stormConf.get(this.clientConfigKey); this.client = new AstyanaxClient<K, C, V>(); this.client.start(config); } CassandraBolt(String clientConfigKey, TupleMapper<K, C, V> tupleMapper); @SuppressWarnings("unchecked") void prepare(Map<String, Object> stormConf, TopologyContext context); void cleanup(); void writeTuple(Tuple input, TupleMapper<K, C, V> tupleMapper); void writeTuples(List<Tuple> inputs, TupleMapper<K, C, V> tupleMapper); Map<String, Object> getComponentConfiguration(); void incrementCounter(Tuple input, TupleCounterMapper tupleMapper); void incrementCounters(List<Tuple> inputs, TupleCounterMapper tupleMapper);  }
@Test public void testBolt() throws Exception { TupleMapper<String, String, String> tupleMapper = new DefaultTupleMapper(KEYSPACE, "users", "VALUE"); String configKey = "cassandra-config"; CassandraBatchingBolt<String, String, String> bolt = new CassandraBatchingBolt<String, String, String>(configKey, tupleMapper); TopologyBuilder builder = new TopologyBuilder(); builder.setBolt("TEST_BOLT", bolt); Fields fields = new Fields("VALUE"); TopologyContext context = new MockTopologyContext(builder.createTopology(), fields); Config config = new Config(); config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 5000); Map<String, Object> clientConfig = new HashMap<String, Object>(); clientConfig.put(StormCassandraConstants.CASSANDRA_HOST, "localhost:9160"); clientConfig.put(StormCassandraConstants.CASSANDRA_KEYSPACE, Arrays.asList(new String [] {KEYSPACE})); config.put(configKey, clientConfig); bolt.prepare(config, context, null); System.out.println("Bolt Preparation Complete."); Values values = new Values(42); Tuple tuple = new TupleImpl(context, values, 5, "test"); bolt.execute(tuple); Thread.sleep(250); AstyanaxContext<Keyspace> astyContext = newContext("localhost:9160", KEYSPACE); Keyspace ks = astyContext.getEntity(); Column<String> result = ks .prepareQuery(new ColumnFamily<String, String>("users", StringSerializer.get(), StringSerializer.get())) .getKey("42").getColumn("VALUE").execute().getResult(); assertEquals("42", result.getStringValue()); }
@Test public void testCounterBolt() throws Exception { String configKey = "cassandra-config"; CassandraCounterBatchingBolt<String, String,Long> bolt = new CassandraCounterBatchingBolt<String, String, Long>(KEYSPACE, configKey, "Counts", "Timestamp", "IncrementAmount"); TopologyBuilder builder = new TopologyBuilder(); builder.setBolt("TEST__COUNTER_BOLT", bolt); Fields fields = new Fields("Timestamp", "IncrementAmount", "CounterColumn"); TopologyContext context = new MockTopologyContext(builder.createTopology(), fields); Config config = new Config(); config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 5000); Map<String, Object> clientConfig = new HashMap<String, Object>(); clientConfig.put(StormCassandraConstants.CASSANDRA_HOST, "localhost:9160"); clientConfig.put(StormCassandraConstants.CASSANDRA_KEYSPACE, Arrays.asList(new String [] {KEYSPACE})); config.put(configKey, clientConfig); bolt.prepare(config, context, null); System.out.println("Bolt Preparation Complete."); Values values = new Values("1", 1L, "MyCountColumn"); Tuple tuple = new TupleImpl(context, values, 5, "test"); bolt.execute(tuple); Thread.sleep(250); AstyanaxContext<Keyspace> astyContext = newContext("localhost:9160", KEYSPACE); Keyspace ks = astyContext.getEntity(); Column<String> result = ks .prepareQuery( new ColumnFamily<String, String>("Counts", StringSerializer.get(), StringSerializer.get())) .getKey("1").getColumn("MyCountColumn").execute().getResult(); assertEquals(1L, result.getLongValue()); }