Add support for a smarter SimpleStoreRecovery and throw Unrecoverable…

…TimeoutException instead of TimeoutException
OpenHFT · May 22, 2016 · c2e7b7f · c2e7b7f
1 parent e032dc4
commit c2e7b7f
Show file tree

Hide file tree

Showing 18 changed files with 854 additions and 835 deletions.
diff --git a/src/main/java/net/openhft/chronicle/queue/ChronicleReader.java b/src/main/java/net/openhft/chronicle/queue/ChronicleReader.java
@@ -27,7 +27,6 @@
 
 import java.io.File;
 import java.io.IOException;
-import java.util.concurrent.TimeoutException;
 
 /**
  * Display records in a Chronicle in a text form.
@@ -55,12 +54,9 @@ public static void tailFileFrom(String basePath, String regex, long index, boole
         if (index > 0) {
             System.out.println("Waiting for index " + index);
             for (; ; ) {
-                try {
-                    if (tailer.moveToIndex(index))
-                        break;
-                } catch (TimeoutException e) {
-                    Jvm.pause(500);
-                }
+                if (tailer.moveToIndex(index))
+                    break;
+                Jvm.pause(100);
             }
         }
 

diff --git a/src/main/java/net/openhft/chronicle/queue/ExcerptAppender.java b/src/main/java/net/openhft/chronicle/queue/ExcerptAppender.java
@@ -17,6 +17,7 @@
 
 import net.openhft.chronicle.bytes.Bytes;
 import net.openhft.chronicle.wire.MarshallableOut;
+import net.openhft.chronicle.wire.UnrecoverableTimeoutException;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.StreamCorruptedException;
@@ -31,7 +32,7 @@ public interface ExcerptAppender extends ExcerptCommon<ExcerptAppender>, Marshal
     /**
      * @param bytes to write to excerpt.
      */
-    void writeBytes(@NotNull Bytes<?> bytes);
+    void writeBytes(@NotNull Bytes<?> bytes) throws UnrecoverableTimeoutException;
 
     /**
      * Write an entry at a given index. This can use used for rebuilding a queue, or replication.

diff --git a/src/main/java/net/openhft/chronicle/queue/ExcerptTailer.java b/src/main/java/net/openhft/chronicle/queue/ExcerptTailer.java
@@ -22,8 +22,6 @@
 import net.openhft.chronicle.wire.SourceContext;
 import org.jetbrains.annotations.NotNull;
 
-import java.util.concurrent.TimeoutException;
-
 /**
  * The component that facilitates sequentially reading data from a {@link ChronicleQueue}.
  *
@@ -64,7 +62,7 @@ default DocumentContext readingDocument() {
      *              with this cycle
      * @return true if this is a valid entries.
      */
-    boolean moveToIndex(long index) throws TimeoutException;
+    boolean moveToIndex(long index);
 
     /**
      * Replay from the first entry in the first cycle.

diff --git a/src/main/java/net/openhft/chronicle/queue/QueueInternal.java b/src/main/java/net/openhft/chronicle/queue/QueueInternal.java
diff --git a/src/main/java/net/openhft/chronicle/queue/impl/WireStore.java b/src/main/java/net/openhft/chronicle/queue/impl/WireStore.java
@@ -18,13 +18,11 @@
 import net.openhft.chronicle.bytes.MappedBytes;
 import net.openhft.chronicle.core.ReferenceCounted;
 import net.openhft.chronicle.queue.impl.single.ScanResult;
-import net.openhft.chronicle.wire.Demarshallable;
-import net.openhft.chronicle.wire.Wire;
-import net.openhft.chronicle.wire.WriteMarshallable;
+import net.openhft.chronicle.wire.*;
 import org.jetbrains.annotations.NotNull;
 
 import java.io.EOFException;
-import java.util.concurrent.TimeoutException;
+import java.io.StreamCorruptedException;
 
 public interface WireStore extends ReferenceCounted, Demarshallable, WriteMarshallable {
     WireStore writePosition(long position);
@@ -40,7 +38,7 @@ public interface WireStore extends ReferenceCounted, Demarshallable, WriteMarsha
      */
     long writePosition();
 
-    ScanResult moveToIndex(@NotNull Wire wire, long index, long timeoutMS) throws TimeoutException;
+    ScanResult moveToIndexForRead(@NotNull Wire wire, long index, long timeoutMS);
 
     @NotNull
     MappedBytes bytes();
@@ -52,13 +50,17 @@ public interface WireStore extends ReferenceCounted, Demarshallable, WriteMarsha
      * @param timeoutMS
      * @return index in this store.
      */
-    long indexForPosition(Wire wire, long position, long timeoutMS) throws EOFException, TimeoutException;
+    long indexForPosition(Wire wire, long position, long timeoutMS) throws EOFException, UnrecoverableTimeoutException, StreamCorruptedException;
 
     String dump();
 
     void lastAcknowledgedIndexReplicated(long lastAcknowledgedIndexReplicated);
 
     long lastAcknowledgedIndexReplicated();
 
-    void setPositionForIndex(Wire wire, long index, long position, long timeoutMS);
+    void setPositionForIndex(Wire wire, long index, long position, long timeoutMS) throws UnrecoverableTimeoutException, StreamCorruptedException;
+
+    long writeHeader(Wire wire, int length, long timeoutMS) throws EOFException, UnrecoverableTimeoutException;
+
+    void writeEOF(AbstractWire wire, long timeoutMS) throws UnrecoverableTimeoutException;
 }