MemoryChannel extends BasicChannelSemantics { @Override public void configure(Context context) { String strCapacity = context.getString("capacity"); Integer capacity = null; if(strCapacity == null) { capacity = defaultCapacity; } else { try { capacity = Integer.parseInt(strCapacity); } catch(NumberFormatException e) { capacity = defaultCapacity; } } String strTransCapacity = context.getString("transactionCapacity"); if(strTransCapacity == null) { transCapacity = defaultTransCapacity; } else { try { transCapacity = Integer.parseInt(strTransCapacity); } catch(NumberFormatException e) { transCapacity = defaultTransCapacity; } } Preconditions.checkState(transCapacity <= capacity); String strKeepAlive = context.getString("keep-alive"); if (strKeepAlive == null) { keepAlive = defaultKeepAlive; } else { keepAlive = Integer.parseInt(strKeepAlive); } if(queue != null) { try { resizeQueue(capacity); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } else { synchronized(queueLock) { queue = new LinkedBlockingDeque<Event>(capacity); queueRemaining = new Semaphore(capacity); queueStored = new Semaphore(0); } } } MemoryChannel(); @Override void configure(Context context);  }
@Test public void testPutTake() throws InterruptedException, EventDeliveryException { Event event = EventBuilder.withBody("test event".getBytes()); Context context = new Context(); Configurables.configure(channel, context); Transaction transaction = channel.getTransaction(); Assert.assertNotNull(transaction); transaction.begin(); channel.put(event); transaction.commit(); transaction.close(); transaction = channel.getTransaction(); Assert.assertNotNull(transaction); transaction.begin(); Event event2 = channel.take(); Assert.assertEquals(event, event2); transaction.commit(); }
@Test public void testChannelResize() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("capacity", "5"); parms.put("transactionCapacity", "5"); context.putAll(parms); Configurables.configure(channel, context); Transaction transaction = channel.getTransaction(); transaction.begin(); for(int i=0; i < 5; i++) { channel.put(EventBuilder.withBody(String.format("test event %d", i).getBytes())); } transaction.commit(); transaction.close(); transaction = channel.getTransaction(); boolean overflowed = false; try { transaction.begin(); channel.put(EventBuilder.withBody("overflow event".getBytes())); transaction.commit(); } catch (ChannelException e) { overflowed = true; transaction.rollback(); } finally { transaction.close(); } Assert.assertTrue(overflowed); parms.put("capacity", "6"); context.putAll(parms); Configurables.configure(channel, context); transaction = channel.getTransaction(); transaction.begin(); channel.put(EventBuilder.withBody("extended capacity event".getBytes())); transaction.commit(); transaction.close(); parms.put("capacity", "2"); parms.put("transactionCapacity", "2"); context.putAll(parms); Configurables.configure(channel, context); for(int i=0; i < 6; i++) { transaction = channel.getTransaction(); transaction.begin(); Assert.assertNotNull(channel.take()); transaction.commit(); transaction.close(); } }
@Test(expected=ChannelException.class) public void testTransactionPutCapacityOverload() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("capacity", "5"); parms.put("transactionCapacity", "2"); context.putAll(parms); Configurables.configure(channel, context); Transaction transaction = channel.getTransaction(); transaction.begin(); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); Assert.fail(); }
@Test(expected=ChannelException.class) public void testCapacityOverload() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("capacity", "5"); parms.put("transactionCapacity", "3"); context.putAll(parms); Configurables.configure(channel, context); Transaction transaction = channel.getTransaction(); transaction.begin(); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); transaction.commit(); transaction.close(); transaction = channel.getTransaction(); transaction.begin(); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); transaction.commit(); Assert.fail(); }
@Test public void testBufferEmptyingAfterTakeCommit() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("capacity", "3"); parms.put("transactionCapacity", "3"); context.putAll(parms); Configurables.configure(channel, context); Transaction tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); tx.commit(); tx.close(); tx = channel.getTransaction(); tx.begin(); channel.take(); channel.take(); tx.commit(); tx.close(); tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); tx.commit(); tx.close(); }
@Test public void testBufferEmptyingAfterRollback() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("capacity", "3"); parms.put("transactionCapacity", "3"); context.putAll(parms); Configurables.configure(channel, context); Transaction tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); tx.rollback(); tx.close(); tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); channel.put(EventBuilder.withBody("test".getBytes())); tx.commit(); tx.close(); }