ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [02/50] [abbrv] incubator-ignite git commit: # IGNITE-141 - S3 fixes
Date Wed, 11 Mar 2015 09:56:27 GMT
# IGNITE-141 - S3 fixes


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/236f40f6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/236f40f6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/236f40f6

Branch: refs/heads/ignite-410
Commit: 236f40f6dca77e6f5d2f3e749ffe14153e04b452
Parents: 116f52f
Author: Valentin Kulichenko <vkulichenko@gridgain.com>
Authored: Thu Mar 5 22:17:44 2015 -0800
Committer: Valentin Kulichenko <vkulichenko@gridgain.com>
Committed: Thu Mar 5 22:17:44 2015 -0800

----------------------------------------------------------------------
 .../spi/checkpoint/s3/S3CheckpointData.java     | 62 ++++++++++++++++++--
 .../spi/checkpoint/s3/S3CheckpointSpi.java      | 14 ++---
 2 files changed, 64 insertions(+), 12 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/236f40f6/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointData.java
----------------------------------------------------------------------
diff --git a/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointData.java
b/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointData.java
index 3f50048..9b18615 100644
--- a/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointData.java
+++ b/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointData.java
@@ -29,10 +29,7 @@ import java.io.*;
  * Host name is used by {@link S3CheckpointSpi} SPI to give node
  * correct files if it is restarted.
  */
-class S3CheckpointData implements Serializable {
-    /** */
-    private static final long serialVersionUID = 0L;
-
+class S3CheckpointData {
     /** Checkpoint data. */
     private final byte[] state;
 
@@ -84,6 +81,63 @@ class S3CheckpointData implements Serializable {
         return key;
     }
 
+    /**
+     * @return Serialized checkpoint data.
+     */
+    public byte[] toBytes() {
+        byte[] keyBytes = key.getBytes();
+
+        byte[] bytes = new byte[4 + state.length + 8 + 4 + keyBytes.length];
+
+        U.intToBytes(state.length, bytes, 0);
+        U.arrayCopy(state, 0, bytes, 4, state.length);
+        U.longToBytes(expTime, bytes, 4 + state.length);
+        U.intToBytes(keyBytes.length, bytes, 4 + state.length + 8);
+        U.arrayCopy(keyBytes, 0, bytes, 4 + state.length + 8 + 4, keyBytes.length);
+
+        return bytes;
+    }
+
+    /**
+     * @param in Input stream.
+     * @return Checkpoint data.
+     * @throws IOException In case of error.
+     */
+    public static S3CheckpointData fromStream(InputStream in) throws IOException {
+        byte[] buf = new byte[8];
+
+        read(in, buf, 4);
+
+        byte[] state = new byte[U.bytesToInt(buf, 0)];
+
+        read(in, state, state.length);
+
+        read(in, buf, 8);
+
+        long expTime = U.bytesToLong(buf, 0);
+
+        read(in, buf, 4);
+
+        byte[] keyBytes = new byte[U.bytesToInt(buf, 0)];
+
+        read(in, keyBytes, keyBytes.length);
+
+        return new S3CheckpointData(state, expTime, new String(keyBytes));
+    }
+
+    /**
+     * @param in Input stream.
+     * @param buf Buffer.
+     * @param len Number of bytes to read.
+     * @throws IOException In case of error.
+     */
+    private static void read(InputStream in, byte[] buf, int len) throws IOException {
+        int cnt = in.read(buf, 0, len);
+
+        if (cnt < len)
+            throw new IOException("End of stream reached.");
+    }
+
     /** {@inheritDoc} */
     @Override public String toString() {
         return S.toString(S3CheckpointData.class, this);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/236f40f6/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointSpi.java
----------------------------------------------------------------------
diff --git a/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointSpi.java
b/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointSpi.java
index 8251d30..9a43fc6 100644
--- a/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointSpi.java
+++ b/modules/aws/src/main/java/org/apache/ignite/spi/checkpoint/s3/S3CheckpointSpi.java
@@ -27,8 +27,6 @@ import org.apache.ignite.internal.util.*;
 import org.apache.ignite.internal.util.tostring.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.marshaller.*;
-import org.apache.ignite.marshaller.jdk.*;
 import org.apache.ignite.resources.*;
 import org.apache.ignite.spi.*;
 import org.apache.ignite.spi.checkpoint.*;
@@ -124,10 +122,6 @@ public class S3CheckpointSpi extends IgniteSpiAdapter implements CheckpointSpi,
     /** Suffix to use in bucket name generation. */
     public static final String DFLT_BUCKET_NAME_SUFFIX = "default-bucket";
 
-    /** Marshaller. */
-    @GridToStringExclude
-    private final Marshaller marsh = new JdkMarshaller();
-
     /** Client to interact with S3 storage. */
     @GridToStringExclude
     private AmazonS3 s3;
@@ -464,7 +458,11 @@ public class S3CheckpointSpi extends IgniteSpiAdapter implements CheckpointSpi,
             InputStream in = obj.getObjectContent();
 
             try {
-                return marsh.unmarshal(in, U.gridClassLoader());
+                return S3CheckpointData.fromStream(in);
+            }
+            catch (IOException e) {
+                throw new IgniteCheckedException("Failed to unmarshal S3CheckpointData [bucketName="
+
+                    bucketName + ", key=" + key + ']', e);
             }
             finally {
                 U.closeQuiet(in);
@@ -492,7 +490,7 @@ public class S3CheckpointSpi extends IgniteSpiAdapter implements CheckpointSpi,
         if (log.isDebugEnabled())
             log.debug("Writing data to S3 [bucket=" + bucketName + ", key=" + data.getKey()
+ ']');
 
-        byte[] buf = marsh.marshal(data);
+        byte[] buf = data.toBytes();
 
         ObjectMetadata meta = new ObjectMetadata();
 


Mime
View raw message