Java 类javax.jms.CompletionListener 实例源码

@Override
public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, CompletionListener listener) throws JMSException {
    checkClosed();

    checkDestinationNotInvalid(destination);

    if (!anonymousProducer) {
        throw new UnsupportedOperationException("Using this method is not supported on producers created with an explicit Destination.");
    }

    if (listener == null) {
        throw new IllegalArgumentException("JmsCompletetionListener cannot be null");
    }

    sendMessage(destination, message, deliveryMode, priority, timeToLive, listener);
}

protected void send(JmsMessageProducer producer, Destination dest, Message msg, int deliveryMode, int priority, long timeToLive, boolean disableMsgId, boolean disableTimestamp, long deliveryDelay, CompletionListener listener) throws JMSException {
    if (dest == null) {
        throw new InvalidDestinationException("Destination must not be null");
    }

    if (msg == null) {
        throw new MessageFormatException("Message must not be null");
    }

    JmsDestination destination = JmsMessageTransformation.transformDestination(connection, dest);

    if (destination.isTemporary() && ((JmsTemporaryDestination) destination).isDeleted()) {
        throw new IllegalStateException("Temporary destination has been deleted");
    }

    send(producer, destination, msg, deliveryMode, priority, timeToLive, disableMsgId, disableTimestamp, deliveryDelay, listener);
}

@Override
public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, CompletionListener listener) throws JMSException {
    checkClosed();

    checkDestinationNotInvalid(destination);

    if (!anonymousProducer) {
        throw new UnsupportedOperationException("Using this method is not supported on producers created with an explicit Destination.");
    }

    if (listener == null) {
        throw new IllegalArgumentException("JmsCompletetionListener cannot be null");
    }

    sendMessage(destination, message, deliveryMode, priority, timeToLive, listener);
}

@Override
public void send(Message message, int deliveryMode, int priority, long timeToLive, CompletionListener listener) throws JMSException {
    checkClosed();

    if (anonymousProducer) {
        throw new UnsupportedOperationException("Using this method is not supported on producers created without an explicit Destination");
    }

    if (listener == null) {
        throw new IllegalArgumentException("JmsCompletetionListener cannot be null");
    }

    sendMessage(destination, message, deliveryMode, priority, timeToLive, listener);
}

private void sendMessage(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, CompletionListener listener) throws JMSException {
    MessageProducer messageProducer = getMessageProducer();

    // Only one thread can use the producer at a time to allow for dynamic configuration
    // changes to match what's been configured here.
    synchronized (messageProducer) {

        long oldDelayValue = 0;
        if (deliveryDelay != 0 && session.isJMSVersionSupported(2, 0)) {
            oldDelayValue = messageProducer.getDeliveryDelay();
            messageProducer.setDeliveryDelay(deliveryDelay);
        }

        // For the non-shared MessageProducer that is also not an anonymous producer we
        // need to call the send method for an explicit MessageProducer otherwise we
        // would be violating the JMS specification in regards to send calls.
        //
        // In all other cases we create an anonymous producer so we call the send with
        // destination parameter version.
        try {
            if (!shared && !anonymousProducer) {
                if (listener == null) {
                    messageProducer.send(message, deliveryMode, priority, timeToLive);
                } else {
                    messageProducer.send(message, deliveryMode, priority, timeToLive, listener);
                }
            } else {
                if (listener == null) {
                    messageProducer.send(destination, message, deliveryMode, priority, timeToLive);
                } else {
                    messageProducer.send(destination, message, deliveryMode, priority, timeToLive, listener);
                }
            }
        } finally {
            if (deliveryDelay != 0 && session.isJMSVersionSupported(2, 0)) {
                messageProducer.setDeliveryDelay(oldDelayValue);
            }
        }
    }
}

@Override
public void send(Message message, CompletionListener completionListener) throws JMSException {
    checkClosed();

    if (anonymousProducer) {
        throw new UnsupportedOperationException("Using this method is not supported on producers created without an explicit Destination");
    }

    if (completionListener == null) {
        throw new IllegalArgumentException("CompletetionListener cannot be null");
    }

    session.send(this, destination, message, deliveryMode, priority, timeToLive, disableMessageId, disableTimestamp, deliveryDelay, completionListener);
}

@Override
public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, CompletionListener completionListener) throws JMSException {
    checkClosed();
    checkDestinationNotInvalid(destination);

    if (!anonymousProducer) {
        throw new UnsupportedOperationException("Using this method is not supported on producers created with an explicit Destination.");
    }

    if (completionListener == null) {
        throw new IllegalArgumentException("CompletionListener cannot be null");
    }

    session.send(this, destination, message, deliveryMode, priority, timeToLive, disableMessageId, disableTimestamp, deliveryMode, null);
}

@Override
public void send(
        final Message message,
        final CompletionListener completionListener
) throws JMSException {
    send(message, 0, 0, 0, completionListener);
}

@Override
public void send(
        final Message message,
        final int deliveryMode,
        final int priority,
        final long timeToLive,
        final CompletionListener completionListener
) throws JMSException {
    messageSink.accept(new ReceivedJmsMessage(destination, message));
    completionListener.onCompletion(message);
}

@Override
public void send(
        final Destination destination,
        final Message message,
        final CompletionListener completionListener
) throws JMSException {
    throw new UnsupportedOperationException("Destination was specified at creation time");
}

@Override
public void send(
        final Destination destination,
        final Message message,
        final int deliveryMode,
        final int priority,
        final long timeToLive,
        final CompletionListener completionListener
) throws JMSException {
    throw new UnsupportedOperationException("Destination was specified at creation time");
}

@Override
public void send(Message message, int deliveryMode, int priority, long timeToLive,
    CompletionListener completionListener) throws JMSException {
  Span span = TracingMessageUtils.buildAndInjectSpan(getDestination(), message, tracer);
  messageProducer.send(message, deliveryMode, priority, timeToLive,
      new TracingCompletionListener(span, completionListener));
}

@Override
public void send(Destination destination, Message message, CompletionListener completionListener)
    throws JMSException {
  Span span = TracingMessageUtils.buildAndInjectSpan(destination, message, tracer);
  messageProducer
      .send(destination, message, new TracingCompletionListener(span, completionListener));
}

@Override
public void send(Destination destination, Message message, int deliveryMode, int priority,
    long timeToLive, CompletionListener completionListener) throws JMSException {
  Span span = TracingMessageUtils.buildAndInjectSpan(destination, message, tracer);
  messageProducer.send(destination, message, deliveryMode, priority, timeToLive,
      new TracingCompletionListener(span, completionListener));
}

@Override
public void send(
    final Destination destination,
    final Message message,
    final int deliveryMode,
    final int priority,
    final long timeToLive,
    final CompletionListener completionListener) throws JMSException {
  if (isClosed()) {
    throw new IllegalStateException("Producer has been closed.");
  }

  if (!getDestination().equals(destination)) {
    throw new IllegalArgumentException("Destination [" + destination
        + "] is invalid. Expected [" + getDestination() + "].");
  }

  final RpcFuture<String> messageIdFuture = publisher.publish(
      PubsubMessage.newBuilder()
          .setData(ByteString.copyFromUtf8(message.getBody(String.class)))
          .build());

  messageIdFuture.addCallback(
      new RpcFutureCallback<String>() {
        @Override public void onSuccess(final String messageId) {
          LOGGER.fine(String.format("%s has been sent successfully.", messageId));
          if (null != completionListener) {
            completionListener.onCompletion(message);
          }
        }

        @Override public void onFailure(final Throwable thrown) {
          LOGGER.log(Level.SEVERE, "Message sending error:", thrown);
          if (null != completionListener) {
            completionListener.onException(message, (Exception) thrown);
          }
        }
      });
}

@Override
public void send(final Message message,
                 final int deliveryMode,
                 final int priority,
                 final long timeToLive,
                 final CompletionListener completionListener) throws JMSException {
  send(destination, message, deliveryMode, priority, timeToLive, completionListener);
}

@Override
public void send(Message message, CompletionListener completionListener) throws JMSException {
   if (ActiveMQRAMessageProducer.trace) {
      ActiveMQRALogger.LOGGER.trace("send(" + message + ", " + completionListener + ")");
   }
   producer.send(message, completionListener);
}

@Override
public void send(Message message,
                 int deliveryMode,
                 int priority,
                 long timeToLive,
                 CompletionListener completionListener) throws JMSException {
   if (ActiveMQRAMessageProducer.trace) {
      ActiveMQRALogger.LOGGER.trace("send(" + message + ", " + deliveryMode + ", " + priority + ", " + timeToLive +
                                       ", " + completionListener + ")");
   }
   producer.send(message, deliveryMode, priority, timeToLive, completionListener);
}

@Override
public void send(Destination destination,
                 Message message,
                 CompletionListener completionListener) throws JMSException {
   if (ActiveMQRAMessageProducer.trace) {
      ActiveMQRALogger.LOGGER.trace("send(" + destination + ", " + message + ", " + completionListener + ")");
   }
   producer.send(destination, message, completionListener);
}

@Override
public void send(Destination destination,
                 Message message,
                 int deliveryMode,
                 int priority,
                 long timeToLive,
                 CompletionListener completionListener) throws JMSException {
   if (ActiveMQRAMessageProducer.trace) {
      ActiveMQRALogger.LOGGER.trace("send(" + destination + ", " + message + ", " + deliveryMode + ", " + priority +
                                       ", " + timeToLive + ", " + completionListener + ")");
   }
   producer.send(destination, message, deliveryMode, priority, timeToLive, completionListener);
}

@Override
public JMSProducer send(Destination destination, Message message) {
   if (message == null) {
      throw new MessageFormatRuntimeException("null message");
   }

   try {
      if (jmsHeaderCorrelationID != null) {
         message.setJMSCorrelationID(jmsHeaderCorrelationID);
      }
      if (jmsHeaderCorrelationIDAsBytes != null && jmsHeaderCorrelationIDAsBytes.length > 0) {
         message.setJMSCorrelationIDAsBytes(jmsHeaderCorrelationIDAsBytes);
      }
      if (jmsHeaderReplyTo != null) {
         message.setJMSReplyTo(jmsHeaderReplyTo);
      }
      if (jmsHeaderType != null) {
         message.setJMSType(jmsHeaderType);
      }
      // XXX HORNETQ-1209 "JMS 2.0" can this be a foreign msg?
      // if so, then "SimpleString" properties will trigger an error.
      setProperties(message);
      if (completionListener != null) {
         CompletionListener wrapped = new CompletionListenerWrapper(completionListener);
         producer.send(destination, message, wrapped);
      } else {
         producer.send(destination, message);
      }
   } catch (JMSException e) {
      throw JmsExceptionUtils.convertToRuntimeException(e);
   }
   return this;
}

@Override
public void send(Message message,
                 int deliveryMode,
                 int priority,
                 long timeToLive,
                 CompletionListener completionListener) throws JMSException {
   checkCompletionListener(completionListener);
   checkDefaultDestination();
   doSendx(defaultDestination, message, deliveryMode, priority, timeToLive, completionListener);
}

@Override
public void send(Destination destination,
                 Message message,
                 int deliveryMode,
                 int priority,
                 long timeToLive,
                 CompletionListener completionListener) throws JMSException {
   checkClosed();

   checkCompletionListener(completionListener);

   checkDestination(destination);

   doSendx((ActiveMQDestination) destination, message, deliveryMode, priority, timeToLive, completionListener);
}

/**
 * @param jmsMessage
 * @param producer
 */
private CompletionListenerWrapper(CompletionListener listener,
                                  Message jmsMessage,
                                  ActiveMQMessageProducer producer) {
   this.completionListener = listener;
   this.jmsMessage = jmsMessage;
   this.producer = producer;
}

@Override
public void send(Message message, int deliveryMode, int priority, long timeToLive, CompletionListener listener) throws JMSException {
    checkClosed();

    if (anonymousProducer) {
        throw new UnsupportedOperationException("Using this method is not supported on producers created without an explicit Destination");
    }

    if (listener == null) {
        throw new IllegalArgumentException("JmsCompletetionListener cannot be null");
    }

    sendMessage(producerInfo.getDestination(), message, deliveryMode, priority, timeToLive, listener);
}

private void sendMessages(int count, JmsMessageProducer producer, CompletionListener listener) throws Exception {
    for (int i = 0; i < count; ++i) {
        Message message = session.createMessage();
        message.setIntProperty("sequence", i);

        producer.send(message, listener);
    }
}

@Override
public void send(Message message, CompletionListener listener) throws JMSException {
    send(message, deliveryMode, priority, timeToLive, listener);
}

@Override
public void send(Destination destination, Message message, CompletionListener listener) throws JMSException {
    send(destination, message, this.deliveryMode, this.priority, this.timeToLive, listener);
}

@Override
public CompletionListener getAsync() {
    return completionListener;
}

@Override
public JMSProducer setAsync(CompletionListener completionListener) {
    this.completionListener = completionListener;
    return this;
}

void send(MockJMSMessageProducer producer, Destination destination, Message message, int deliveryMode, int priority, long timeToLive, boolean disableMessageId, boolean disableTimestamp, long deliveryDelay, CompletionListener completionListener) throws JMSException {
    sendLock.lock();
    try {
        message.setJMSDeliveryMode(deliveryMode);
        message.setJMSPriority(priority);
        message.setJMSRedelivered(false);
        message.setJMSDestination(destination);

        long timeStamp = System.currentTimeMillis();
        boolean hasTTL = timeToLive > Message.DEFAULT_TIME_TO_LIVE;
        boolean hasDelay = deliveryDelay > Message.DEFAULT_DELIVERY_DELAY;

        if (!(message instanceof MockJMSMessage)) {
            throw new IllegalStateException("Mock JMS client cannot handle foreign messages");
        }

        if (!disableTimestamp) {
            message.setJMSTimestamp(timeStamp);
        } else {
            message.setJMSTimestamp(0);
        }

        if (hasTTL) {
            message.setJMSExpiration(timeStamp + timeToLive);
        } else {
            message.setJMSExpiration(0);
        }

        long messageSequence = producer.getNextMessageSequence();
        String messageId = null;
        if (!disableMessageId) {
            messageId = producer.getProducerId() + ":"+ messageSequence;
        }

        // Set the delivery time. Purposefully avoided doing this earlier so
        // that we use the 'outbound' JmsMessage object reference when
        // updating our own message instances, avoids using the interface
        // in case the JMS 1.1 Message API is actually being used due to
        // being on the classpath too.
        long deliveryTime = timeStamp;
        if (hasDelay) {
            deliveryTime = timeStamp + deliveryDelay;
        }

        message.setJMSDeliveryTime(deliveryTime);

        // Set the message ID
        message.setJMSMessageID(messageId);

        try {
            connection.onMessageSend(this, message);
        } catch (JMSException jmsEx) {
            // If the synchronous portion of the send fails the completion be
            // notified but might depending on the circumstances of the failures,
            // remove it from the queue and check if is is already completed
            // once we decide to add completion support to the mock
            throw jmsEx;
        }
    } finally {
        sendLock.unlock();
    }
}

@Override
public void send(Destination destination, Message message, CompletionListener completionListener) throws JMSException {
    send(message, deliveryMode, priority, timeToLive, completionListener);
}

@Override
public void send(Message message, int deliveryMode, int priority, long timeToLive, CompletionListener completionListener) throws JMSException {
    send(destination, message, this.deliveryMode, this.priority, this.timeToLive, completionListener);
}

@Override
public CompletionListener getAsync() {
    return completionListener;
}

@Override
public JMSProducer setAsync(CompletionListener completionListener) {
    this.completionListener = completionListener;
    return this;
}

@Override
public void send(Message arg0, CompletionListener arg1)
        throws JMSException {
    // TODO Auto-generated method stub

}

@Override
public void send(Destination arg0, Message arg1, CompletionListener arg2)
        throws JMSException {
    // TODO Auto-generated method stub

}

@Override
public void send(Message arg0, int arg1, int arg2, long arg3,
        CompletionListener arg4) throws JMSException {
    // TODO Auto-generated method stub

}

@Override
public void send(Destination arg0, Message arg1, int arg2, int arg3,
        long arg4, CompletionListener arg5) throws JMSException {
    // TODO Auto-generated method stub

}

public TracingCompletionListener(Span span, CompletionListener completionListener) {
  this.span = span;
  this.completionListener = completionListener;
}