From 577620533d6a39ba196a69b4464616e1e6b5da5f Mon Sep 17 00:00:00 2001
From: jbertram <jbertram@apache.org>
Date: Fri, 22 Jul 2016 11:12:40 -0500
Subject: [PATCH] Fix ByteBuffer regression from Netty upgrade

Using array() is a bit dangerous as it's an optional part of any
ByteBuffer implementation. This new method will deal with various
ByteBuffer implementations appropriately.
---
 .../apache/activemq/artemis/utils/ByteUtil.java    | 14 ++++++++++++++
 .../core/client/impl/ClientConsumerImpl.java       |  3 ++-
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java
index 50ac3cc5b6..79210721a1 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/ByteUtil.java
@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.artemis.utils;
 
+import java.nio.ByteBuffer;
+
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
@@ -90,4 +92,16 @@ public class ByteUtil {
       return sb.toString();
    }
 
+   public static byte[] getActiveArray(ByteBuffer buffer) {
+      byte[] ret = new byte[buffer.remaining()];
+      if (buffer.hasArray()) {
+         byte[] array = buffer.array();
+         System.arraycopy(array, buffer.arrayOffset() + buffer.position(), ret, 0, ret.length);
+      }
+      else {
+         buffer.slice().get(ret);
+      }
+      return ret;
+   }
+
 }
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientConsumerImpl.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientConsumerImpl.java
index 57bb869bb6..42168e9f61 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientConsumerImpl.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientConsumerImpl.java
@@ -38,6 +38,7 @@ import org.apache.activemq.artemis.core.client.ActiveMQClientLogger;
 import org.apache.activemq.artemis.core.client.ActiveMQClientMessageBundle;
 import org.apache.activemq.artemis.spi.core.remoting.ConsumerContext;
 import org.apache.activemq.artemis.spi.core.remoting.SessionContext;
+import org.apache.activemq.artemis.utils.ByteUtil;
 import org.apache.activemq.artemis.utils.FutureLatch;
 import org.apache.activemq.artemis.utils.PriorityLinkedList;
 import org.apache.activemq.artemis.utils.PriorityLinkedListImpl;
@@ -640,7 +641,7 @@ public final class ClientConsumerImpl implements ClientConsumerInternal {
       //sets the packet
       ActiveMQBuffer qbuff = clMessage.getBodyBuffer();
       int bytesToRead = qbuff.writerIndex() - qbuff.readerIndex();
-      final byte[] body = qbuff.readBytes(bytesToRead).toByteBuffer().array();
+      final byte[] body = ByteUtil.getActiveArray(qbuff.readBytes(bytesToRead).toByteBuffer());
 
       largeMessage.setLargeMessageController(new CompressedLargeMessageControllerImpl(currentLargeMessageController));
       currentLargeMessageController.addPacket(body, body.length, false);