RecoverableMemoryChannel extends BasicChannelSemantics { @Override public synchronized void start() { LOG.info("Starting " + this); try { WALReplayResult<RecoverableMemoryChannelEvent> results = wal.replay(); Preconditions.checkArgument(results.getSequenceID() >= 0); LOG.info("Replay SequenceID " + results.getSequenceID()); seqidGenerator.set(results.getSequenceID()); int numResults = results.getResults().size(); Preconditions.checkState(numResults <= capacity, "Capacity " + capacity + ", but we need to replay " + numResults); LOG.info("Replay Events " + numResults); for(WALEntry<RecoverableMemoryChannelEvent> entry : results.getResults()) { seqidGenerator.set(Math.max(entry.getSequenceID(),seqidGenerator.get())); } for(WALEntry<RecoverableMemoryChannelEvent> entry : results.getResults()) { Transaction transaction = null; try { transaction = memoryChannel.getTransaction(); transaction.begin(); memoryChannel.put(entry.getData()); transaction.commit(); } catch(Exception e) { if(transaction != null) { try { transaction.rollback(); } catch(Exception ex) { LOG.info("Error during rollback", ex); } } Throwables.propagate(e); } catch(Error e) { if(transaction != null) { try { transaction.rollback(); } catch(Exception ex) { LOG.info("Error during rollback", ex); } } throw e; } finally { if(transaction != null) { transaction.close(); } } } } catch (IOException e) { Throwables.propagate(e); } super.start(); open = true; } RecoverableMemoryChannel(); @Override void configure(Context context); @Override synchronized void start(); @Override synchronized void stop(); static final String WAL_DATA_DIR; static final String WAL_ROLL_SIZE; static final String WAL_MAX_LOGS_SIZE; static final String WAL_MIN_RENTENTION_PERIOD; static final String WAL_WORKER_INTERVAL; static final String CAPACITY; static final String KEEPALIVE; static final int DEFAULT_CAPACITY; static final int DEFAULT_KEEPALIVE; }
@Test public void testThreaded() throws IOException, InterruptedException { int numThreads = 10; final CountDownLatch producerStopLatch = new CountDownLatch(numThreads); final CountDownLatch consumerStartLatch = new CountDownLatch(numThreads); final CountDownLatch consumerStopLatch = new CountDownLatch(numThreads); final List<Exception> errors = Collections .synchronizedList(new ArrayList<Exception>()); final List<String> expected = Collections .synchronizedList(new ArrayList<String>()); final List<String> actual = Collections .synchronizedList(new ArrayList<String>()); for (int i = 0; i < numThreads; i++) { final int id = i; Thread t = new Thread() { @Override public void run() { try { consumerStartLatch.await(); if (id % 2 == 0) { expected.addAll(putEvents(channel, Integer.toString(id), 1, 5)); } else { expected.addAll(putEvents(channel, Integer.toString(id), 5, 5)); } logger.info("Completed some puts " + expected.size()); } catch (Exception e) { logger.error("Error doing puts", e); errors.add(e); } finally { producerStopLatch.countDown(); } } }; t.setDaemon(true); t.start(); } for (int i = 0; i < numThreads; i++) { final int id = i; Thread t = new Thread() { @Override public void run() { try { consumerStartLatch.countDown(); consumerStartLatch.await(); while(!producerStopLatch.await(1, TimeUnit.SECONDS) || expected.size() > actual.size()) { if (id % 2 == 0) { actual.addAll(takeEvents(channel, 1, Integer.MAX_VALUE)); } else { actual.addAll(takeEvents(channel, 5, Integer.MAX_VALUE)); } } if(actual.isEmpty()) { logger.error("Found nothing!"); } else { logger.info("Completed some takes " + actual.size()); } } catch (Exception e) { logger.error("Error doing takes", e); errors.add(e); } finally { consumerStopLatch.countDown(); } } }; t.setDaemon(true); t.start(); } Assert.assertTrue("Timed out waiting for producers", producerStopLatch.await(30, TimeUnit.SECONDS)); Assert.assertTrue("Timed out waiting for consumer", consumerStopLatch.await(30, TimeUnit.SECONDS)); Assert.assertEquals(Collections.EMPTY_LIST, errors); Collections.sort(expected); Collections.sort(actual); Assert.assertEquals(expected, actual); }