@Test public void testClose() throws IOException { File bd = mkdir("close"); PersistentMessageBuffer mb = new PersistentMessageBuffer(bd); assertTrue(mb.isOpen()); mb.close(); mb.close(); assertFalse(mb.isOpen()); try { mb.append(0L, "", new byte[0]); fail(); } catch (IOException e) { } }
@Override public void close() throws IOException { Cursor[] copyOfWaitingCursors; synchronized (this) { if (!isOpen()) return; if (syncTask != null) { syncTask.cancel(); syncTask = null; } if (current != null) { current.close(); current = null; } open = false; copyOfWaitingCursors = waitingCursors; } for (Cursor c : copyOfWaitingCursors) { if (c != null) c.interrupt(); } }
PersistentMessageBuffer implements MessageBuffer { @Override public void close() throws IOException { Cursor[] copyOfWaitingCursors; synchronized (this) { if (!isOpen()) return; if (syncTask != null) { syncTask.cancel(); syncTask = null; } if (current != null) { current.close(); current = null; } open = false; copyOfWaitingCursors = waitingCursors; } for (Cursor c : copyOfWaitingCursors) { if (c != null) c.interrupt(); } } }
PersistentMessageBuffer implements MessageBuffer { @Override public void close() throws IOException { Cursor[] copyOfWaitingCursors; synchronized (this) { if (!isOpen()) return; if (syncTask != null) { syncTask.cancel(); syncTask = null; } if (current != null) { current.close(); current = null; } open = false; copyOfWaitingCursors = waitingCursors; } for (Cursor c : copyOfWaitingCursors) { if (c != null) c.interrupt(); } } PersistentMessageBuffer(File dir); }
PersistentMessageBuffer implements MessageBuffer { @Override public void close() throws IOException { Cursor[] copyOfWaitingCursors; synchronized (this) { if (!isOpen()) return; if (syncTask != null) { syncTask.cancel(); syncTask = null; } if (current != null) { current.close(); current = null; } open = false; copyOfWaitingCursors = waitingCursors; } for (Cursor c : copyOfWaitingCursors) { if (c != null) c.interrupt(); } } PersistentMessageBuffer(File dir); @Override synchronized void setFirstId(long firstMessageId); @Override void setMaxSize(long bytes); @Override long getMaxSize(); void setSegmentCount(int segmentCount); int getSegmentCount(); @Override void setMaxPayloadSize(int maxPayloadSize); @Override int getMaxPayloadSize(); void setSegmentLength(int segmentLength); int getSegmentLength(); @Override void setExecutor(Executor executor); @Override synchronized boolean isEmpty(); @Override synchronized long getSize(); synchronized int getFileCount(); @Override long append(long timestamp, String routingKey, byte[] payload); @SuppressWarnings({"ConstantConditions", "SynchronizationOnLocalVariableOrMethodParameter"}) @Override long append(long timestamp, String routingKey, ReadableByteChannel payload, int payloadSize); @Override int getMessageSize(String routingKey, int payloadSize); @Override void setAutoSyncInterval(int ms); @Override int getAutoSyncInterval(); @Override void setTimer(Timer timer); void cleanup(); @Override synchronized void sync(); @Override synchronized boolean isOpen(); @Override void close(); @Override synchronized long getNextId(); @Override synchronized long getMessageCount(); @Override synchronized Date getOldestTimestamp(); @Override synchronized long getOldestId(); synchronized Date getMostRecentTimestamp(); @Override long getCreationTime(); @Override synchronized Timeline getTimeline(); @Override synchronized Timeline getTimeline(long messageId); @Override String toString(); @Override synchronized MessageCursor cursor(long messageId); @Override synchronized MessageCursor cursorByTimestamp(long timestamp); }
PersistentMessageBuffer implements MessageBuffer { @Override public void close() throws IOException { Cursor[] copyOfWaitingCursors; synchronized (this) { if (!isOpen()) return; if (syncTask != null) { syncTask.cancel(); syncTask = null; } if (current != null) { current.close(); current = null; } open = false; copyOfWaitingCursors = waitingCursors; } for (Cursor c : copyOfWaitingCursors) { if (c != null) c.interrupt(); } } PersistentMessageBuffer(File dir); @Override synchronized void setFirstId(long firstMessageId); @Override void setMaxSize(long bytes); @Override long getMaxSize(); void setSegmentCount(int segmentCount); int getSegmentCount(); @Override void setMaxPayloadSize(int maxPayloadSize); @Override int getMaxPayloadSize(); void setSegmentLength(int segmentLength); int getSegmentLength(); @Override void setExecutor(Executor executor); @Override synchronized boolean isEmpty(); @Override synchronized long getSize(); synchronized int getFileCount(); @Override long append(long timestamp, String routingKey, byte[] payload); @SuppressWarnings({"ConstantConditions", "SynchronizationOnLocalVariableOrMethodParameter"}) @Override long append(long timestamp, String routingKey, ReadableByteChannel payload, int payloadSize); @Override int getMessageSize(String routingKey, int payloadSize); @Override void setAutoSyncInterval(int ms); @Override int getAutoSyncInterval(); @Override void setTimer(Timer timer); void cleanup(); @Override synchronized void sync(); @Override synchronized boolean isOpen(); @Override void close(); @Override synchronized long getNextId(); @Override synchronized long getMessageCount(); @Override synchronized Date getOldestTimestamp(); @Override synchronized long getOldestId(); synchronized Date getMostRecentTimestamp(); @Override long getCreationTime(); @Override synchronized Timeline getTimeline(); @Override synchronized Timeline getTimeline(long messageId); @Override String toString(); @Override synchronized MessageCursor cursor(long messageId); @Override synchronized MessageCursor cursorByTimestamp(long timestamp);  }