import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Override
protected void doWrite(Message<?> message) throws Exception {
	Object payload = message.getPayload();
	if (payload instanceof String) {
		String data = (String)payload;
		if (delimiter != null) {
			processor.onNext(Buffer.wrap(data+delimiter));
		} else {
			processor.onNext(Buffer.wrap(data));
		}
		if (meter != null) {
			if ((meterCount++ % rateInterval) == 0) {
				meter.mark(rateInterval);
				log.info("METER: 1 minute rate = " + meter.getOneMinuteRate() + " mean rate = " + meter.getMeanRate());
			}
		}
	} else {
		throw new MessageHandlingException(message, "message not a String");
	}
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
protected Object handleRequestMessage(final Message<?> requestMessage) {
	
	if (!running) {
		return null;
	}
	
	Future<Object> response = executorService.submit(new Callable<Object>() {
		public Object call() throws Exception {
			byte[] requestData = requestConverter.convert(requestMessage.getPayload());
			socket.send(requestData);
			byte[] replyData = socket.recv();
			if (replyData == null) {
				socket.close();
				ZmqOutboundGateway.this.connect();
				throw ZmqEndpointUtil.buildMessageHandlingException(requestMessage, socket.base().errno());
			}
			return replyConverter.convert(replyData);
		}
	});
			
	try {
		return response.get();
	} catch (Throwable t) {
		throw new MessageHandlingException(requestMessage, t);
	}
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void verifyHeadersAreJustWrong() throws IOException {
  // assemble
  Client mockClient = mockClientFactory.getClient();
  Queue queue = mock(Queue.class);
  when(mockClient.queue("queue3")).thenReturn(queue);

  // the message sender does it's best to convert any header value
  String timeout = "x";

  // act
  c3.send(MessageBuilder
    .withPayload("message3")
    .setHeader("ironmq_queue", "queue3")
    .setHeader(IronMqMessageHeaders.TIMEOUT, timeout)
    .build());

  // assert
  verify(mockClient).queue("queue3");
  verify(queue, times(0)).push(anyString(), anyLong());
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void testEvaluationIncorrectTupleInput() {
	Tuple incompleteInputTuple = TupleBuilder.tuple()
			//	missing data type
			.put(TF_SHAPE, new long[0])
			.put(TF_VALUE, new byte[0])
			.build();
	testEvaluation(incompleteInputTuple);
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message)
		throws Exception {
	Principal user = ((WampMessage) message).getPrincipal();
	if (user == null) {
		throw new MessageHandlingException(message,
				"No \"PRINCIPAL\" header in message");
	}
	return user;
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void missingPrincipalTest() throws Exception {
	CallMessage callMessage = new CallMessage(1, "call");
	TestPrincipal testPrincipal = new TestPrincipal("testPrincipal");
	assertThat(this.resolver.resolveArgument(this.principalParameter, callMessage))
			.isEqualTo(testPrincipal);
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void testEvaluationIncorrectTupleInput() throws InterruptedException {
	Tuple incompleteInputTuple = TupleBuilder.tuple()
			//	missing data type
			.put(TF_SHAPE, new long[0])
			.put(TF_VALUE, new byte[0])
			.build();
	testEvaluation(incompleteInputTuple);
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Override
protected final void handleMessageInternal(Message<?> message) throws Exception {
	try {
		doWrite(message);
	}
	catch (Exception e) {
		throw new MessageHandlingException(message,
				"failed to write Message payload to GPDB/HAWQ", e);
	}
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
private void doHandleMessage(MessageHandler handler, Message<?> message) {
	try {
		handler.handleMessage(message);
	}
	catch (Throwable t) { // NOSONAR
		throw new MessageHandlingException(message, t);
	}
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
@Test(expected = MessageHandlingException.class)
public void testStreamListenerJavaSerializationNonSerializable() throws Exception {
	Binder binder = getBinder();

	BindingProperties producerBindingProperties = createProducerBindingProperties(createProducerProperties());

	DirectChannel moduleOutputChannel = createBindableChannel("output", producerBindingProperties);

	BindingProperties consumerBindingProperties = createConsumerBindingProperties(createConsumerProperties());

	DirectChannel moduleInputChannel = createBindableChannel("input", consumerBindingProperties);

	Binding<MessageChannel> producerBinding = binder.bindProducer(String.format("bad%s0c",
			getDestinationNameDelimiter()), moduleOutputChannel, producerBindingProperties.getProducer());

	Binding<MessageChannel> consumerBinding = binder.bindConsumer(String.format("bad%s0c",
			getDestinationNameDelimiter()), "test-3", moduleInputChannel, consumerBindingProperties.getConsumer());
	try {
		Station station = new Station();
		Message<?> message = MessageBuilder.withPayload(station)
				.setHeader(MessageHeaders.CONTENT_TYPE,
						MessageConverterUtils.X_JAVA_SERIALIZED_OBJECT)
				.build();
		moduleOutputChannel.send(message);
	}
	finally {
		producerBinding.unbind();
		consumerBinding.unbind();
	}
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message)
		throws Exception {
	Principal user = ((WampMessage) message).getPrincipal();

	if (user == null) {
		throw new MessageHandlingException(message,
				"No \"principal\" header in message");
	}

	return user;
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message)
		throws Exception {
	WampSession wampSession = ((WampMessage) message).getWampSession();

	if (wampSession == null) {
		throw new MessageHandlingException(message,
				"No \"wampSession\" header in message");
	}

	return wampSession;
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void missingPrincipalTest() throws Exception {
	CallMessage callMessage = new CallMessage("1", "call");
	TestPrincipal testPrincipal = new TestPrincipal("testPrincipal");
	assertThat(this.resolver.resolveArgument(this.principalParameter, callMessage))
			.isEqualTo(testPrincipal);
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void missingWampSessionTest() throws Exception {
	CallMessage callMessage = new CallMessage("1", "call");
	WampSession wampSession = new WampSession(this.nativeSession);
	assertThat(this.resolver.resolveArgument(this.wampSessionParameter, callMessage))
			.isEqualTo(wampSession);
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
protected Object handleRequestMessage(final Message<?> requestMessage) {
	if (!running) {
		return null;
	}
	
	Future<Object> response = executorService.submit(new Callable<Object>() {
		public Object call() throws Exception {
			byte[] requestData = requestConverter.convert(requestMessage.getPayload());
			int retriesLeft = retryCount;
			while (!Thread.currentThread().isInterrupted()) {
				socket.send(requestData);
				PollItem items[] = { new PollItem(socket, Poller.POLLIN) };
				int rc = ZMQ.poll(items, socketReceiveTimeout);
				if (rc == -1) {
					break;
				}
				if (items[0].isReadable()) {
					byte[] reply = socket.recv();
					return replyConverter.convert(reply);
				} else if (--retriesLeft == 0) {
					break;
				} else {
					ZmqLazyPirateGateway.this.connect();
				}
			}
			ZmqLazyPirateGateway.this.connect();
			return null;
		}
	});
			
	try {
		return response.get();
	} catch (Throwable t) {
		throw new MessageHandlingException(requestMessage, t);
	}
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
public static MessageHandlingException buildMessageHandlingException(Message<?> failedMessage, int errno)  {
	if (errno == ZError.ETERM) {
		return new MessageHandlingException(failedMessage, "ZMQ Context has been terminated");
	} else if (errno == ZError.EAGAIN) {
		return new MessageHandlingException(failedMessage, "ZMQ socket timeout out while attempting to send or receive data over zmq socket");
	} else {
		return new MessageHandlingException(failedMessage, "Unknown error while attempting to send or receive data over zmq socket");
	}
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Override
protected void handleMissingValue(String name, MethodParameter parameter, Message<?> message) {
	throw new MessageHandlingException(message, "Missing path template variable '" + name +
			"' for method parameter type [" + parameter.getParameterType() + "]");
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Override
protected void handleMissingValue(String headerName, MethodParameter parameter, Message<?> message) {
	throw new MessageHandlingException(message, "Missing header '" + headerName +
			"' for method parameter type [" + parameter.getParameterType() + "]");
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void resolveArgumentNotFound() throws Exception {
	Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).build();
	this.resolver.resolveArgument(this.paramRequired, message);
}
 

import org.springframework.messaging.MessageHandlingException; //导入依赖的package包/类
@Test(expected = MessageHandlingException.class)
public void resolveArgumentNotFound() throws Exception {
	Message<byte[]> message = MessageBuilder.withPayload(new byte[0]).build();
	this.resolver.resolveArgument(this.paramAnnotated, message);
}