GEODE-6687: replace ByteArrayInputStream+DataInputStream with ByteArr…

…ayDataInput
b-xiang · May 1, 2019 · 8c2c9c8 · 8c2c9c8
1 parent 1603ea0
commit 8c2c9c8
Show file tree

Hide file tree

Showing 18 changed files with 47 additions and 73 deletions.
diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AbstractOp.java
@@ -15,15 +15,14 @@
 
 package org.apache.geode.cache.client.internal;
 
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
 import java.net.SocketTimeoutException;
 
 import org.apache.logging.log4j.Logger;
 
 import org.apache.geode.InternalGemFireError;
 import org.apache.geode.cache.client.ServerConnectivityException;
 import org.apache.geode.cache.client.ServerOperationException;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.HeapDataOutputStream;
 import org.apache.geode.internal.Version;
 import org.apache.geode.internal.cache.PutAllPartialResultException;
@@ -159,7 +158,7 @@ protected void processSecureBytes(Connection cnx, Message message) throws Except
         return;
       }
       byte[] bytes = ((ConnectionImpl) cnx).decryptBytes(partBytes);
-      DataInputStream dis = new DataInputStream(new ByteArrayInputStream(bytes));
+      ByteArrayDataInput dis = new ByteArrayDataInput(bytes);
       cnx.setConnectionID(dis.readLong());
     }
   }

diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/AuthenticateUserOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/AuthenticateUserOp.java
@@ -17,8 +17,6 @@
 
 import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT;
 
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
 import java.util.Properties;
 
 import org.apache.geode.DataSerializer;
@@ -29,6 +27,7 @@
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.distributed.internal.ServerLocation;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.HeapDataOutputStream;
 import org.apache.geode.internal.Version;
 import org.apache.geode.internal.cache.tier.MessageType;
@@ -182,7 +181,7 @@ protected Object processResponse(Message msg, Connection cnx) throws Exception {
         } else {
           cnx.getServer().setRequiresCredentials(true);
           byte[] decrypted = ((ConnectionImpl) cnx).decryptBytes(bytes);
-          DataInputStream dis = new DataInputStream(new ByteArrayInputStream(decrypted));
+          ByteArrayDataInput dis = new ByteArrayDataInput(decrypted);
           userId = dis.readLong();
         }
         if (needsServerLocation) {

diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClientSideHandshakeImpl.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/ClientSideHandshakeImpl.java
@@ -17,7 +17,6 @@
 import static org.apache.geode.distributed.ConfigurationProperties.CONFLATE_EVENTS;
 import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT;
 
-import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.DataOutputStream;
@@ -50,6 +49,7 @@
 import org.apache.geode.distributed.internal.LonerDistributionManager;
 import org.apache.geode.distributed.internal.ServerLocation;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.HeapDataOutputStream;
 import org.apache.geode.internal.InternalDataSerializer;
 import org.apache.geode.internal.InternalInstantiator;
@@ -269,12 +269,8 @@ public ServerQueueStatus handshakeWithServer(Connection conn, ServerLocation loc
   private InternalDistributedMember readServerMember(DataInputStream p_dis) throws IOException {
 
     byte[] memberBytes = DataSerializer.readByteArray(p_dis);
-    ByteArrayInputStream bais = new ByteArrayInputStream(memberBytes);
-    DataInputStream dis = new DataInputStream(bais);
     Version v = InternalDataSerializer.getVersionForDataStreamOrNull(p_dis);
-    if (v != null) {
-      dis = new VersionedDataInputStream(dis, v);
-    }
+    ByteArrayDataInput dis = new ByteArrayDataInput(memberBytes, v);
     try {
       return DataSerializer.readObject(dis);
     } catch (EOFException e) {

diff --git a/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutOp.java b/geode-core/src/main/java/org/apache/geode/cache/client/internal/PutOp.java
@@ -15,8 +15,6 @@
 
 package org.apache.geode.cache.client.internal;
 
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
 
 import org.apache.logging.log4j.Logger;
 
@@ -29,6 +27,7 @@
 import org.apache.geode.cache.client.ServerOperationException;
 import org.apache.geode.distributed.internal.ServerLocation;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.cache.CachedDeserializable;
 import org.apache.geode.internal.cache.EntryEventImpl;
 import org.apache.geode.internal.cache.LocalRegion;
@@ -290,8 +289,7 @@ protected Object processResponse(Message msg, Connection con) throws Exception {
         if ((flags & HAS_OLD_VALUE_FLAG) != 0) {
           oldValue = msg.getPart(partIdx++).getObject();
           if ((flags & OLD_VALUE_IS_OBJECT_FLAG) != 0 && oldValue instanceof byte[]) {
-            ByteArrayInputStream in = new ByteArrayInputStream((byte[]) oldValue);
-            DataInputStream din = new DataInputStream(in);
+            ByteArrayDataInput din = new ByteArrayDataInput((byte[]) oldValue);
             oldValue = DataSerializer.readObject(din);
           }
         }

diff --git a/geode-core/src/main/java/org/apache/geode/cache/wan/EventSequenceID.java b/geode-core/src/main/java/org/apache/geode/cache/wan/EventSequenceID.java
@@ -14,11 +14,10 @@
  */
 package org.apache.geode.cache.wan;
 
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
 import java.util.Arrays;
 
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.ByteArrayDataInput;
 
 /**
  * This class wraps 1) DistributedMembershipID 2) ThreadID 3) SequenceID attributes which are used
@@ -48,7 +47,7 @@ public EventSequenceID(byte[] membershipID, long threadID, long sequenceID) {
     Object mbr;
     try {
       mbr = InternalDistributedMember
-          .readEssentialData(new DataInputStream(new ByteArrayInputStream(membershipID)));
+          .readEssentialData(new ByteArrayDataInput(membershipID));
     } catch (Exception e) {
       mbr = Arrays.toString(membershipID); // punt and use the bytes
     }

diff --git a/geode-core/src/main/java/org/apache/geode/internal/ByteArrayDataInput.java b/geode-core/src/main/java/org/apache/geode/internal/ByteArrayDataInput.java
@@ -42,6 +42,14 @@ public class ByteArrayDataInput extends InputStream implements DataInput, Versio
    */
   public ByteArrayDataInput() {}
 
+  public ByteArrayDataInput(byte[] bytes) {
+    initialize(bytes, null);
+  }
+
+  public ByteArrayDataInput(byte[] bytes, Version version) {
+    initialize(bytes, version);
+  }
+
   /**
    * Initialize this byte array stream with given byte array and version.
    *

diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/EntryEventImpl.java b/geode-core/src/main/java/org/apache/geode/internal/cache/EntryEventImpl.java
@@ -17,9 +17,7 @@
 import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_NEW_VALUE;
 import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.ENTRY_EVENT_OLD_VALUE;
 
-import java.io.ByteArrayInputStream;
 import java.io.DataInput;
-import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.function.Function;
@@ -1810,7 +1808,7 @@ private void processDeltaBytes(Object oldValueInVM) {
       try {
         long start = CachePerfStats.getStatTime();
         ((org.apache.geode.Delta) value)
-            .fromDelta(new DataInputStream(new ByteArrayInputStream(getDeltaBytes())));
+            .fromDelta(new ByteArrayDataInput(getDeltaBytes()));
         getRegion().getCachePerfStats().endDeltaUpdate(start);
         deltaBytesApplied = true;
       } catch (RuntimeException rte) {

diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/EventID.java b/geode-core/src/main/java/org/apache/geode/internal/cache/EventID.java
@@ -14,9 +14,7 @@
  */
 package org.apache.geode.internal.cache;
 
-import java.io.ByteArrayInputStream;
 import java.io.DataInput;
-import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.Externalizable;
 import java.io.IOException;
@@ -39,11 +37,11 @@
 import org.apache.geode.distributed.DistributedSystem;
 import org.apache.geode.distributed.internal.DistributionConfig;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.DataSerializableFixedID;
 import org.apache.geode.internal.HeapDataOutputStream;
 import org.apache.geode.internal.InternalDataSerializer;
 import org.apache.geode.internal.Version;
-import org.apache.geode.internal.VersionedDataInputStream;
 import org.apache.geode.internal.cache.ha.HARegionQueue;
 import org.apache.geode.internal.cache.ha.ThreadIdentifier;
 import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
@@ -310,13 +308,13 @@ public InternalDistributedMember getDistributedMember() {
    * have UUID bytes in the memberID. Newer clients don't require this.
    */
   public InternalDistributedMember getDistributedMember(Version targetVersion) {
-    ByteArrayInputStream bais = new ByteArrayInputStream(this.membershipID);
-    DataInputStream dis = new DataInputStream(bais);
+    Version disVersion = null;
     if (targetVersion.compareTo(Version.GEODE_1_1_0) < 0) {
       // GEODE-3153: clients expect to receive UUID bytes, which are only
       // read if the stream's version is 1.0.0-incubating
-      dis = new VersionedDataInputStream(dis, Version.GFE_90);
+      disVersion = Version.GFE_90;
     }
+    ByteArrayDataInput dis = new ByteArrayDataInput(membershipID, disVersion);
     InternalDistributedMember result = null;
     try {
       result = InternalDistributedMember.readEssentialData(dis);
@@ -506,7 +504,7 @@ public String expensiveToString() {
     Object mbr;
     try {
       mbr = InternalDistributedMember
-          .readEssentialData(new DataInputStream(new ByteArrayInputStream(membershipID)));
+          .readEssentialData(new ByteArrayDataInput(membershipID));
     } catch (Exception e) {
       mbr = membershipID; // punt and use the bytes
     }

diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/FilterRoutingInfo.java b/geode-core/src/main/java/org/apache/geode/internal/cache/FilterRoutingInfo.java
@@ -14,12 +14,9 @@
  */
 package org.apache.geode.internal.cache;
 
-import java.io.ByteArrayInputStream;
 import java.io.DataInput;
-import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.IOException;
-import java.io.InputStream;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
@@ -32,11 +29,11 @@
 import org.apache.geode.distributed.DistributedMember;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.HeapDataOutputStream;
 import org.apache.geode.internal.InternalDataSerializer;
 import org.apache.geode.internal.ObjToByteArraySerializer;
 import org.apache.geode.internal.Version;
-import org.apache.geode.internal.VersionedDataInputStream;
 import org.apache.geode.internal.VersionedDataSerializable;
 
 /**
@@ -490,13 +487,7 @@ public void setInterestedClientsInv(Set clients) {
      */
     private void deserialize() {
       try {
-        InputStream is = new ByteArrayInputStream(myData);
-        DataInputStream dis;
-        if (this.myDataVersion != null) {
-          dis = new VersionedDataInputStream(is, this.myDataVersion);
-        } else {
-          dis = new DataInputStream(is);
-        }
+        ByteArrayDataInput dis = new ByteArrayDataInput(myData, myDataVersion);
         boolean hasCQs = dis.readBoolean();
         if (hasCQs) {
           int numEntries = InternalDataSerializer.readArrayLength(dis);

diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/ha/ThreadIdentifier.java b/geode-core/src/main/java/org/apache/geode/internal/cache/ha/ThreadIdentifier.java
@@ -14,16 +14,15 @@
  */
 package org.apache.geode.internal.cache.ha;
 
-import java.io.ByteArrayInputStream;
 import java.io.DataInput;
-import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.IOException;
 import java.util.Arrays;
 
 import org.apache.geode.DataSerializable;
 import org.apache.geode.DataSerializer;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.cache.EventID;
 
 /**
@@ -204,7 +203,7 @@ public String expensiveToString() {
     Object mbr;
     try {
       mbr = InternalDistributedMember
-          .readEssentialData(new DataInputStream(new ByteArrayInputStream(membershipID)));
+          .readEssentialData(new ByteArrayDataInput(membershipID));
     } catch (Exception e) {
       mbr = membershipID; // punt and use the bytes
     }

diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AuthIds.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/AuthIds.java
@@ -16,15 +16,15 @@
 
 package org.apache.geode.internal.cache.tier.sockets;
 
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
+
+import org.apache.geode.internal.ByteArrayDataInput;
 
 public class AuthIds {
   private long connectionId;
   private long uniqueId;
 
   public AuthIds(byte[] bytes) throws Exception {
-    DataInputStream dis = new DataInputStream(new ByteArrayInputStream(bytes));
+    ByteArrayDataInput dis = new ByteArrayDataInput(bytes);
     if (bytes.length == 8) {
       // only connectionid
       connectionId = dis.readLong();

diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxy.java b/geode-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientProxy.java
@@ -14,8 +14,6 @@
  */
 package org.apache.geode.internal.cache.tier.sockets;
 
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
 import java.io.IOException;
 import java.net.Socket;
 import java.net.SocketException;
@@ -70,6 +68,7 @@
 import org.apache.geode.distributed.internal.ClusterDistributionManager;
 import org.apache.geode.distributed.internal.DistributionConfig;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.SystemTimer;
 import org.apache.geode.internal.SystemTimer.SystemTimerTask;
 import org.apache.geode.internal.Version;
@@ -2852,7 +2851,7 @@ protected Object deserialize(byte[] serializedBytes) {
       // This is a debugging method so ignore all exceptions like
       // ClassNotFoundException
       try {
-        DataInputStream dis = new DataInputStream(new ByteArrayInputStream(serializedBytes));
+        ByteArrayDataInput dis = new ByteArrayDataInput(serializedBytes);
         deserializedObject = DataSerializer.readObject(dis);
       } catch (Exception e) {
       }

diff --git a/...e-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java b/...e-core/src/main/java/org/apache/geode/internal/cache/tier/sockets/CacheClientUpdater.java
@@ -14,8 +14,6 @@
  */
 package org.apache.geode.internal.cache.tier.sockets;
 
-import java.io.ByteArrayInputStream;
-import java.io.DataInputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InterruptedIOException;
@@ -65,6 +63,7 @@
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.distributed.internal.membership.MemberAttributes;
 import org.apache.geode.internal.Assert;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.InternalDataSerializer;
 import org.apache.geode.internal.InternalInstantiator;
 import org.apache.geode.internal.Version;
@@ -1804,7 +1803,7 @@ private Object deserialize(byte[] serializedBytes) {
     Object deserializedObject = serializedBytes;
     // This is a debugging method so ignore all exceptions like ClassNotFoundException
     try {
-      DataInputStream dis = new DataInputStream(new ByteArrayInputStream(serializedBytes));
+      ByteArrayDataInput dis = new ByteArrayDataInput(serializedBytes);
       deserializedObject = DataSerializer.readObject(dis);
     } catch (ClassNotFoundException | IOException ignore) {
     }

diff --git a/...e/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProxyMembershipID.java b/...e/src/main/java/org/apache/geode/internal/cache/tier/sockets/ClientProxyMembershipID.java
@@ -16,9 +16,7 @@
 
 import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_ID;
 
-import java.io.ByteArrayInputStream;
 import java.io.DataInput;
-import java.io.DataInputStream;
 import java.io.DataOutput;
 import java.io.Externalizable;
 import java.io.IOException;
@@ -38,10 +36,10 @@
 import org.apache.geode.distributed.internal.DistributionConfig;
 import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
 import org.apache.geode.internal.Assert;
+import org.apache.geode.internal.ByteArrayDataInput;
 import org.apache.geode.internal.DataSerializableFixedID;
 import org.apache.geode.internal.HeapDataOutputStream;
 import org.apache.geode.internal.Version;
-import org.apache.geode.internal.VersionedDataInputStream;
 import org.apache.geode.internal.logging.LogService;
 
 /**
@@ -393,10 +391,9 @@ private ClientProxyMembershipID canonicalReference() {
 
   public DistributedMember getDistributedMember() {
     if (memberId == null) {
-      ByteArrayInputStream bais = new ByteArrayInputStream(identity);
-      DataInputStream dis = new VersionedDataInputStream(bais, Version.CURRENT);
+      ByteArrayDataInput dataInput = new ByteArrayDataInput(identity);
       try {
-        memberId = (DistributedMember) DataSerializer.readObject(dis);
+        memberId = (DistributedMember) DataSerializer.readObject(dataInput);
       } catch (Exception e) {
         logger.error("Unable to deserialize membership id", e);
       }