聊聊artemis的SessionProducerCreditsMessage

序

本文主要研究一下artemis的SessionProducerCreditsMessage

SessionProducerCreditsMessage

activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/wireformat/SessionProducerCreditsMessage.java

public class SessionProducerCreditsMessage extends PacketImpl {

   private int credits;

   private SimpleString address;

   public SessionProducerCreditsMessage(final int credits, final SimpleString address) {
      super(SESS_PRODUCER_CREDITS);

      this.credits = credits;

      this.address = address;
   }

   public SessionProducerCreditsMessage() {
      super(SESS_PRODUCER_CREDITS);
   }

   public int getCredits() {
      return credits;
   }

   public SimpleString getAddress() {
      return address;
   }

   @Override
   public void encodeRest(final ActiveMQBuffer buffer) {
      buffer.writeInt(credits);
      buffer.writeSimpleString(address);
   }

   @Override
   public void decodeRest(final ActiveMQBuffer buffer) {
      credits = buffer.readInt();
      address = buffer.readSimpleString();
   }

   @Override
   public int hashCode() {
      final int prime = 31;
      int result = super.hashCode();
      result = prime * result + ((address == null) ? 0 : address.hashCode());
      result = prime * result + credits;
      return result;
   }

   @Override
   public String toString() {
      StringBuffer buff = new StringBuffer(getParentString());
      buff.append(", address=" + address);
      buff.append(", credits=" + credits);
      buff.append("]");
      return buff.toString();
   }

   @Override
   public boolean equals(Object obj) {
      if (this == obj)
         return true;
      if (!super.equals(obj))
         return false;
      if (!(obj instanceof SessionProducerCreditsMessage))
         return false;
      SessionProducerCreditsMessage other = (SessionProducerCreditsMessage) obj;
      if (address == null) {
         if (other.address != null)
            return false;
      } else if (!address.equals(other.address))
         return false;
      if (credits != other.credits)
         return false;
      return true;
   }
}

• SessionProducerCreditsMessage继承了PacketImpl，其type为SESS_PRODUCER_CREDITS

ClientSessionPacketHandler

activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java

class ClientSessionPacketHandler implements ChannelHandler {

      @Override
      public void handlePacket(final Packet packet) {
         byte type = packet.getType();

         try {
            switch (type) {
               case DISCONNECT_CONSUMER: {
                  handleConsumerDisconnected((DisconnectConsumerMessage) packet);
                  break;
               }
               case SESS_RECEIVE_CONTINUATION: {
                  handleReceiveContinuation((SessionReceiveContinuationMessage) packet);

                  break;
               }
               case SESS_RECEIVE_MSG: {
                  handleReceivedMessagePacket((SessionReceiveMessage) packet);

                  break;
               }
               case SESS_RECEIVE_LARGE_MSG: {
                  handleReceiveLargeMessage((SessionReceiveLargeMessage) packet);

                  break;
               }
               case PacketImpl.SESS_PRODUCER_CREDITS: {
                  handleReceiveProducerCredits((SessionProducerCreditsMessage) packet);

                  break;
               }
               case PacketImpl.SESS_PRODUCER_FAIL_CREDITS: {
                  handleReceiveProducerFailCredits((SessionProducerCreditsFailMessage) packet);

                  break;
               }
               case PacketImpl.DISCONNECT_CONSUMER_KILL: {
                  handleReceiveSlowConsumerKillMessage((DisconnectConsumerWithKillMessage) packet);

                  break;
               }
               case EXCEPTION: {
                  // We can only log these exceptions
                  // maybe we should cache it on SessionContext and throw an exception on any next calls
                  ActiveMQExceptionMessage mem = (ActiveMQExceptionMessage) packet;

                  ActiveMQClientLogger.LOGGER.receivedExceptionAsynchronously(mem.getException());

                  break;
               }
               default: {
                  throw ActiveMQClientMessageBundle.BUNDLE.invalidPacket(type);
               }
            }
         } catch (Exception e) {
            throw ActiveMQClientMessageBundle.BUNDLE.failedToHandlePacket(e);
         }

         sessionChannel.confirm(packet);
      }
   }

• ClientSessionPacketHandler实现了ChannelHandler接口，其handlePacket根据不同的type做不同的处理，当type为SESS_PRODUCER_CREDITS时执行handleReceiveProducerCredits方法

ActiveMQSessionContext

activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/protocol/core/impl/ActiveMQSessionContext.java

public class ActiveMQSessionContext extends SessionContext {

   //......

   protected void handleReceiveProducerCredits(SessionProducerCreditsMessage message) {
      handleReceiveProducerCredits(message.getAddress(), message.getCredits());
   }

   //......

}

• handleReceiveProducerCredits调用的是父类SessionContext的handleReceiveProducerCredits方法

SessionContext

activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/artemis/spi/core/remoting/SessionContext.java

public abstract class SessionContext {

   protected ClientSessionInternal session;

   protected SendAcknowledgementHandler sendAckHandler;

   protected volatile RemotingConnection remotingConnection;

   protected final IDGenerator idGenerator = new SimpleIDGenerator(0);

   //......

   protected void handleReceiveProducerCredits(SimpleString address, int credits) {
      ClientSessionInternal session = this.session;
      if (session != null) {
         session.handleReceiveProducerCredits(address, credits);
      }

   }

   //......
}

• handleReceiveProducerCredits方法执行的是ClientSessionInternal的handleReceiveProducerCredits方法

ClientSessionImpl

activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/client/impl/ClientSessionImpl.java

public final class ClientSessionImpl implements ClientSessionInternal, FailureListener {

   //......

   @Override
   public void handleReceiveProducerCredits(final SimpleString address, final int credits) {
      synchronized (producerCreditManager) {
         producerCreditManager.receiveCredits(address, credits);
      }
   }

   //......
}

• handleReceiveProducerCredits调用的是producerCreditManager.receiveCredits方法

小结

SessionProducerCreditsMessage继承了PacketImpl，其type为SESS_PRODUCER_CREDITS；ClientSessionPacketHandler实现了ChannelHandler接口，其handlePacket根据不同的type做不同的处理，当type为SESS_PRODUCER_CREDITS时执行handleReceiveProducerCredits方法；handleReceiveProducerCredits调用的是父类SessionContext的handleReceiveProducerCredits方法；SessionContext的handleReceiveProducerCredits方法执行的是ClientSessionInternal的handleReceiveProducerCredits方法；ClientSessionImpl的handleReceiveProducerCredits调用的是producerCreditManager.receiveCredits方法

doc

• SessionProducerCreditsMessage