import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
protected void initMocks() throws ActiveMQException {
    CoreQueueConfiguration q = new CoreQueueConfiguration();
    q.setName("q");
    ClientSession sesh = mock(ClientSession.class);
    
    producer = mock(ClientProducer.class);
    when(sesh.createProducer(q.getAddress())).thenReturn(producer);
    
    msgToQueue = mock(ClientMessage.class);
    boolean durable = true;
    when(sesh.createMessage(durable)).thenReturn(msgToQueue);
    
    msgBody = mock(ActiveMQBuffer.class);
    when(msgToQueue.getBodyBuffer()).thenReturn(msgBody);
    
    connector = new ArtemisQConnector(q, sesh);
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public synchronized void broadcastConnectors() throws Exception {
   ActiveMQBuffer buff = ActiveMQBuffers.dynamicBuffer(4096);

   buff.writeString(nodeManager.getNodeId().toString());

   buff.writeString(uniqueID);

   buff.writeInt(connectors.size());

   for (TransportConfiguration tcConfig : connectors) {
      tcConfig.encode(buff);
   }

   // Only send as many bytes as we need.
   byte[] data = new byte[buff.readableBytes()];
   buff.getBytes(buff.readerIndex(), data);

   endpoint.broadcast(data);
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public void encode(final ActiveMQBuffer buffer) {
   buffer.writeLong(transactionID);

   boolean isLargeMessage = isLargeMessage();

   buffer.writeBoolean(isLargeMessage);

   if (isLargeMessage) {
      buffer.writeInt(LargeMessagePersister.getInstance().getEncodeSize((LargeServerMessage)message));
      LargeMessagePersister.getInstance().encode(buffer, (LargeServerMessage) message);
   } else {
      message.getPersister().encode(buffer, message);
   }

   buffer.writeInt(queueIDs.length);

   for (long queueID : queueIDs) {
      buffer.writeLong(queueID);
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public void failedTransaction(final long transactionID,
                              final List<RecordInfo> records,
                              final List<RecordInfo> recordsToDelete) {
   for (RecordInfo record : records) {
      if (record.userRecordType == ADD_LARGE_MESSAGE) {
         byte[] data = record.data;

         ActiveMQBuffer buff = ActiveMQBuffers.wrappedBuffer(data);

         try {
            LargeServerMessage serverMessage = journalStorageManager.parseLargeMessage(messages, buff);
            serverMessage.decrementDelayDeletionCount();
         } catch (Exception e) {
            ActiveMQServerLogger.LOGGER.journalError(e);
         }
      }
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Test
public void testBadProtocol() throws Exception {
   BufferHandler handler = new BufferHandler() {
      @Override
      public void bufferReceived(final Object connectionID, final ActiveMQBuffer buffer) {
      }
   };
   Map<String, Object> params = new HashMap<>();
   params.put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
   params.put(TransportConstants.ENABLED_PROTOCOLS_PROP_NAME, "myBadProtocol");

   NettyConnector connector = new NettyConnector(params, handler, listener, Executors.newCachedThreadPool(ActiveMQThreadFactory.defaultThreadFactory()), Executors.newCachedThreadPool(ActiveMQThreadFactory.defaultThreadFactory()), Executors.newScheduledThreadPool(5, ActiveMQThreadFactory.defaultThreadFactory()));

   connector.start();
   Assert.assertTrue(connector.isStarted());
   Assert.assertNull(connector.createConnection());
   connector.close();
   Assert.assertFalse(connector.isStarted());
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
   buffer.writeString(nodeID);
   buffer.writeNullableString(backupGroupName);
   buffer.writeBoolean(backup);
   buffer.writeLong(currentEventID);
   if (connector != null) {
      buffer.writeBoolean(true);
      connector.encode(buffer);
   } else {
      buffer.writeBoolean(false);
   }
   if (backupConnector != null) {
      buffer.writeBoolean(true);
      backupConnector.encode(buffer);
   } else {
      buffer.writeBoolean(false);
   }
   buffer.writeNullableString(scaleDownGroupName);
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public void bufferReceived(final Object connectionID, final ActiveMQBuffer buffer) {
   ConnectionEntry conn = connections.get(connectionID);

   if (conn != null) {
      try {
         conn.connection.bufferReceived(connectionID, buffer);
      } catch (RuntimeException e) {
         ActiveMQServerLogger.LOGGER.disconnectCritical("Error decoding buffer", e);
         conn.connection.fail(new ActiveMQException(e.getMessage()));
      }
   } else {
      if (logger.isTraceEnabled()) {
         logger.trace("ConnectionID = " + connectionID + " was already closed, so ignoring packet");
      }
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public Packet decode(final ActiveMQBuffer in, CoreRemotingConnection connection) {
   final byte packetType = in.readByte();
   //optimized for the most common cases: hottest and commons methods will be inlined and this::decode too due to the byte code size
   switch (packetType) {
      case SESS_SEND:
         return decodeSessionSendMessage(in, connection);
      case SESS_ACKNOWLEDGE:
         return decodeSessionAcknowledgeMessage(in, connection);
      case SESS_PRODUCER_REQUEST_CREDITS:
         return decodeRequestProducerCreditsMessage(in, connection);
      case SESS_FLOWTOKEN:
         return decodeSessionConsumerFlowCreditMessage(in, connection);
      default:
         return slowPathDecode(in, packetType, connection);
   }
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
/**
 * Utility for reading bytes out of streaming.
 * It will return remainingBytes, bytesRead
 *
 * @param remainingBytes remaining Bytes from previous read. Send it to 0 if it was the first call for the message
 * @param buff
 * @return a pair of remaining bytes and bytes read
 */
public static Pair<Integer, Integer> streamReadBytes(ActiveMQBuffer buff, int remainingBytes, byte[] value) {
   if (remainingBytes == -1) {
      return new Pair<>(0, -1);
   } else if (remainingBytes == 0) {
      byte type = buff.readByte();
      if (type != DataConstants.BYTES) {
         throw new IllegalStateException("Invalid conversion");
      }
      remainingBytes = buff.readInt();
   }
   int read = Math.min(value.length, remainingBytes);
   buff.readBytes(value, 0, read);
   remainingBytes -= read;
   if (remainingBytes == 0) {
      remainingBytes = -1;
   }
   return new Pair<>(remainingBytes, read);

}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Test
public void passMessageToClientProducer() throws Exception {
    ClientMessage msg = mock(ClientMessage.class);
    ArtemisMessage adapter = new ArtemisMessage(msg);
    when(mockMsgFactory.durableMessage()).thenReturn(adapter);
    when(msg.getBodyBuffer()).thenReturn(mock(ActiveMQBuffer.class));

    target.sendMessage(QMsgFactory::durableMessage, out -> {});
    verify(mockProducer, times(1)).send(any());
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Test
public void onMessageForwardsToMessageHandler() {
    ClientMessage msg = mock(ClientMessage.class);
    when(msg.getBodyBuffer()).thenReturn(mock(ActiveMQBuffer.class));

    target.onMessage(msg);

    assertNotNull(receivedMessage);
    assertNotNull(receivedBody);
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
private void mockMsgToRead(int serializedValueLen, byte[] serializedValue) {
    ActiveMQBuffer buf = msgMock.getBodyBuffer();
    when(buf.readInt()).thenReturn(serializedValueLen);
    doAnswer(invocation -> {
        byte[] passedInBuffer = (byte[])invocation.getArguments()[0];
        System.arraycopy(serializedValue, 0,
                         passedInBuffer, 0, passedInBuffer.length);
        return null;
    }).when(buf).readBytes((byte[])any());
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Test
public void receiveMessage() throws Exception {
    String msgData = "msg";
    DequeueTask<ArtemisMessage, String> task = newTask(msgData);
    ClientMessage qMsg = mock(ClientMessage.class);
    ActiveMQBuffer buf = mock(ActiveMQBuffer.class);
    when(qMsg.getBodyBuffer()).thenReturn(buf);

    ArtemisQConsumer qConsumer = (ArtemisQConsumer) task.receiver();
    qConsumer.onMessage(qMsg);
    
    assertThat(receivedMsg.message(), is(qMsg));
    assertThat(receivedData, is(msgData));
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public Map<String, Object> getFields(MessageReference ref) throws OpenDataException {
   Map<String, Object> rc = super.getFields(ref);
   ICoreMessage m = ref.getMessage().toCore();
   if (!m.isLargeMessage()) {
      ActiveMQBuffer bodyCopy = m.getReadOnlyBodyBuffer();
      byte[] bytes = new byte[bodyCopy.readableBytes()];
      bodyCopy.readBytes(bytes);
      rc.put(CompositeDataConstants.BODY, bytes);
   } else {
      rc.put(CompositeDataConstants.BODY, new byte[0]);
   }
   return rc;
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
public static String readLine(ActiveMQBuffer buffer) {
   StringBuilder sb = new StringBuilder("");
   char c = buffer.readChar();
   while (c != '\n') {
      sb.append(c);
      c = buffer.readChar();
   }
   return sb.toString();
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
/**
 * @param id
 * @param buffer
 * @return
 */
protected static QueueStatusEncoding newQueueStatusEncoding(long id, ActiveMQBuffer buffer) {
   QueueStatusEncoding statusEncoding = new QueueStatusEncoding();

   statusEncoding.decode(buffer);
   statusEncoding.setId(id);

   return statusEncoding;
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public void encodeRest(final ActiveMQBuffer buffer) {
   buffer.writeSimpleString(address);
   buffer.writeInt(routingTypes.size());
   for (RoutingType routingType : routingTypes) {
      buffer.writeByte(routingType.getType());
   }
   buffer.writeBoolean(requiresResponse);
   buffer.writeBoolean(autoCreated);
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public Message decode(ActiveMQBuffer buffer, Message record) {
   long id = buffer.readLong();
   long format = buffer.readLong();
   SimpleString address = buffer.readNullableSimpleString();
   record = new AMQPMessage(format);
   record.reloadPersistence(buffer);
   record.setMessageID(id);
   if (address != null) {
      record.setAddress(address);
   }
   return record;
}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
public static int bytesReadBytes(ActiveMQBuffer message, final byte[] value, final int length) {
   if (!message.readable()) {
      return -1;
   }

   int read = Math.min(length, message.readableBytes());

   if (read != 0) {
      message.readBytes(value, 0, read);
   }

   return read;

}
 

import org.apache.activemq.artemis.api.core.ActiveMQBuffer; //导入依赖的package包/类
@Override
public final void write(ActiveMQBuffer buffer,
                        final boolean flush,
                        final boolean batched,
                        final ChannelFutureListener futureListener) {
   final int readableBytes = buffer.readableBytes();
   if (logger.isDebugEnabled()) {
      final int remainingBytes = this.writeBufferHighWaterMark - readableBytes;
      if (remainingBytes < 0) {
         logger.debug("a write request is exceeding by " + (-remainingBytes) + " bytes the writeBufferHighWaterMark size [ " + this.writeBufferHighWaterMark + " ] : consider to set it at least of " + readableBytes + " bytes");
      }
   }
   //no need to lock because the Netty's channel is thread-safe
   //and the order of write is ensured by the order of the write calls
   final EventLoop eventLoop = channel.eventLoop();
   final boolean inEventLoop = eventLoop.inEventLoop();
   if (!inEventLoop) {
      writeNotInEventLoop(buffer, flush, batched, futureListener);
   } else {
      // OLD COMMENT:
      // create a task which will be picked up by the eventloop and trigger the write.
      // This is mainly needed as this method is triggered by different threads for the same channel.
      // if we not do this we may produce out of order writes.
      // NOTE:
      // the submitted task does not effect in any way the current written size in the batch
      // until the loop will process it, leading to a longer life for the ActiveMQBuffer buffer!!!
      // To solve it, will be necessary to manually perform the count of the current batch instead of rely on the
      // Channel:Config::writeBufferHighWaterMark value.
      this.pendingWritesOnEventLoop += readableBytes;
      this.pendingWritesOnEventLoopView.lazySet(pendingWritesOnEventLoop);
      eventLoop.execute(() -> {
         this.pendingWritesOnEventLoop -= readableBytes;
         this.pendingWritesOnEventLoopView.lazySet(pendingWritesOnEventLoop);
         writeInEventLoop(buffer, flush, batched, futureListener);
      });
   }
}