@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(); }
@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 extends BasicChannelSemantics { @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 extends BasicChannelSemantics { @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(); }
MemoryChannel extends BasicChannelSemantics { @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(); }
MemoryChannel extends BasicChannelSemantics { @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();  }