MemoryChannel extends BasicChannelSemantics implements TransactionCapacitySupported { @Override public void configure(Context context) { Integer capacity = null; try { capacity = context.getInteger("capacity", defaultCapacity); } catch (NumberFormatException e) { capacity = defaultCapacity; LOGGER.warn("Invalid capacity specified, initializing channel to " + "default capacity of {}", defaultCapacity); } if (capacity <= 0) { capacity = defaultCapacity; LOGGER.warn("Invalid capacity specified, initializing channel to " + "default capacity of {}", defaultCapacity); } try { transCapacity = context.getInteger("transactionCapacity", defaultTransCapacity); } catch (NumberFormatException e) { transCapacity = defaultTransCapacity; LOGGER.warn("Invalid transation capacity specified, initializing channel" + " to default capacity of {}", defaultTransCapacity); } if (transCapacity <= 0) { transCapacity = defaultTransCapacity; LOGGER.warn("Invalid transation capacity specified, initializing channel" + " to default capacity of {}", defaultTransCapacity); } Preconditions.checkState(transCapacity <= capacity, "Transaction Capacity of Memory Channel cannot be higher than " + "the capacity."); try { byteCapacityBufferPercentage = context.getInteger("byteCapacityBufferPercentage", defaultByteCapacityBufferPercentage); } catch (NumberFormatException e) { byteCapacityBufferPercentage = defaultByteCapacityBufferPercentage; } try { byteCapacity = (int) ((context.getLong("byteCapacity", defaultByteCapacity).longValue() * (1 - byteCapacityBufferPercentage * .01)) / byteCapacitySlotSize); if (byteCapacity < 1) { byteCapacity = Integer.MAX_VALUE; } } catch (NumberFormatException e) { byteCapacity = (int) ((defaultByteCapacity * (1 - byteCapacityBufferPercentage * .01)) / byteCapacitySlotSize); } try { keepAlive = context.getInteger("keep-alive", defaultKeepAlive); } catch (NumberFormatException e) { keepAlive = defaultKeepAlive; } 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); } } if (bytesRemaining == null) { bytesRemaining = new Semaphore(byteCapacity); lastByteCapacity = byteCapacity; } else { if (byteCapacity > lastByteCapacity) { bytesRemaining.release(byteCapacity - lastByteCapacity); lastByteCapacity = byteCapacity; } else { try { if (!bytesRemaining.tryAcquire(lastByteCapacity - byteCapacity, keepAlive, TimeUnit.SECONDS)) { LOGGER.warn("Couldn't acquire permits to downsize the byte capacity, resizing has been aborted"); } else { lastByteCapacity = byteCapacity; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } if (channelCounter == null) { channelCounter = new ChannelCounter(getName()); } } MemoryChannel(); @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); long getTransactionCapacity();  }
@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 testPutAcceptsNullValueInHeader() { Configurables.configure(channel, new Context()); Event event = EventBuilder.withBody("test body".getBytes(Charsets.UTF_8), Collections.<String, String>singletonMap("test_key", null)); Transaction txPut = channel.getTransaction(); txPut.begin(); channel.put(event); txPut.commit(); txPut.close(); Transaction txTake = channel.getTransaction(); txTake.begin(); Event eventTaken = channel.take(); Assert.assertEquals(event, eventTaken); txTake.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 testCapacityBufferEmptyingAfterTakeCommit() { 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 testCapacityBufferEmptyingAfterRollback() { 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(); }
@Test(expected = ChannelException.class) public void testByteCapacityOverload() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("byteCapacity", "2000"); parms.put("byteCapacityBufferPercentage", "20"); context.putAll(parms); Configurables.configure(channel, context); byte[] eventBody = new byte[405]; Transaction transaction = channel.getTransaction(); transaction.begin(); channel.put(EventBuilder.withBody(eventBody)); channel.put(EventBuilder.withBody(eventBody)); channel.put(EventBuilder.withBody(eventBody)); transaction.commit(); transaction.close(); transaction = channel.getTransaction(); transaction.begin(); channel.put(EventBuilder.withBody(eventBody)); channel.put(EventBuilder.withBody(eventBody)); transaction.commit(); Assert.fail(); }
@Test public void testByteCapacityBufferEmptyingAfterRollback() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("byteCapacity", "2000"); parms.put("byteCapacityBufferPercentage", "20"); context.putAll(parms); Configurables.configure(channel, context); byte[] eventBody = new byte[405]; Transaction tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody(eventBody)); channel.put(EventBuilder.withBody(eventBody)); channel.put(EventBuilder.withBody(eventBody)); tx.rollback(); tx.close(); tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody(eventBody)); channel.put(EventBuilder.withBody(eventBody)); channel.put(EventBuilder.withBody(eventBody)); tx.commit(); tx.close(); }
@Test public void testNullEmptyEvent() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("byteCapacity", "2000"); parms.put("byteCapacityBufferPercentage", "20"); context.putAll(parms); Configurables.configure(channel, context); Transaction tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody(null)); tx.commit(); tx.close(); tx = channel.getTransaction(); tx.begin(); channel.put(EventBuilder.withBody(new byte[0])); tx.commit(); tx.close(); }
@Test public void testNegativeCapacities() { Context context = new Context(); Map<String, String> parms = new HashMap<String, String>(); parms.put("capacity", "-3"); parms.put("transactionCapacity", "-1"); context.putAll(parms); Configurables.configure(channel, context); Assert.assertTrue(field("queue") .ofType(LinkedBlockingDeque.class) .in(channel).get() .remainingCapacity() > 0); Assert.assertTrue(field("transCapacity") .ofType(Integer.class) .in(channel).get() > 0); }