dubbo-rpc-extensions/dubbo-rpc-rocketmq/src/test/java/org/apache/dubbo/rpc/rocketmq/RocketMQProtocolTest.java - dubbo-spi-extensions - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.dubbo.rpc.rocketmq;

 import org.apache.dubbo.common.URL;
 import org.apache.dubbo.common.URLBuilder;
 import org.apache.dubbo.common.constants.CommonConstants;
 import org.apache.dubbo.remoting.Channel;
 import org.apache.dubbo.remoting.buffer.ChannelBuffer;
 import org.apache.dubbo.remoting.exchange.Request;
 import org.apache.dubbo.remoting.exchange.Response;
 import org.apache.dubbo.rpc.Exporter;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
 import org.apache.dubbo.rpc.ProtocolServer;
 import org.apache.dubbo.rpc.Result;
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.protocol.AbstractProtocol;
 import org.apache.dubbo.rpc.rocketmq.codec.RocketMQCountCodec;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.MessageSelector;
 import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.common.message.MessageAccessor;
 import org.apache.rocketmq.common.message.MessageConst;
 import org.apache.rocketmq.common.message.MessageExt;

 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;


 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.core.classloader.annotations.PrepareForTest;
 import org.powermock.modules.junit4.PowerMockRunner;
 import org.powermock.reflect.Whitebox;

 @RunWith(PowerMockRunner.class)
 @PrepareForTest(value = {RocketMQProtocol.class})
 public class RocketMQProtocolTest {

     private final RocketMQProtocol pocketMQProtocol = PowerMockito.spy(new RocketMQProtocol());

     private final DefaultMQProducer defaultMQProducer = PowerMockito.mock(DefaultMQProducer.class);

     private final RocketMQProtocolServer rocketMQProtocolServer = PowerMockito.mock(RocketMQProtocolServer.class);

     private final RocketMQExporter<Object> rocketMQExporter = PowerMockito.mock(RocketMQExporter.class);

     private final RocketMQCountCodec codec = PowerMockito.mock(RocketMQCountCodec.class);
     private final List<MessageExt> msgs = new ArrayList<>();
     private final MessageExt messageExt = new MessageExt();
     private MessageListenerConcurrently messageListenerConcurrently;
     private Response response;

     private URL url;


     @Before
     public void before() throws Exception {
         String urlString =
             "nameservice://localhost:9876/org.apache.dubbo.registry.RegistryService?application=rocketmq-provider&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=8990&release=3.0.7&route=false";
         url = URLBuilder.valueOf(urlString).addParameter("groupModel", "topic").addParameter(CommonConstants.ENABLE_TIMEOUT_COUNTDOWN_KEY, false);

         PowerMockito.whenNew(RocketMQProtocolServer.class).withAnyArguments().thenReturn(rocketMQProtocolServer);
         PowerMockito.when(rocketMQProtocolServer.getDefaultMQProducer()).thenReturn(defaultMQProducer);
         PowerMockito.whenNew(RocketMQExporter.class).withAnyArguments().thenReturn(rocketMQExporter);

         PowerMockito.whenNew(RocketMQCountCodec.class).withAnyArguments().thenReturn(codec);
         PowerMockito.doAnswer(new Answer() {
             @Override
             public Object answer(InvocationOnMock invocation)  {
                 messageListenerConcurrently = invocation.getArgument(0);
                 messageListenerConcurrently = PowerMockito.spy(messageListenerConcurrently);
                 return null;
             }
         }).when(rocketMQProtocolServer).setMessageListenerConcurrently(Mockito.any(MessageListenerConcurrently.class));
         PowerMockito.when(pocketMQProtocol, "createServer", Mockito.any(URL.class), Mockito.anyString(), Mockito.anyString()).thenCallRealMethod();

         messageExt.setTopic("test");
         messageExt.putUserProperty(CommonConstants.TIMEOUT_KEY, (System.currentTimeMillis() + 100000) + "");
         messageExt.putUserProperty(RocketMQProtocolConstant.URL_STRING, urlString);
         messageExt.setBody(new byte[1024]);

         msgs.add(messageExt);

     }

     @Test
     public void createServerTest() throws Exception {
         PowerMockito.when(pocketMQProtocol, "createServer", url, "", "").thenCallRealMethod();

         Mockito.verify(rocketMQProtocolServer, Mockito.atLeastOnce()).setModel(Mockito.any(String.class));
         Mockito.verify(rocketMQProtocolServer, Mockito.atLeastOnce()).setMessageListenerConcurrently(Mockito.any(MessageListenerConcurrently.class));
         Mockito.verify(rocketMQProtocolServer, Mockito.atLeastOnce()).reset(url);

     }

     @Test
     public void openServerTest() throws Exception {
         Field serverMapField = AbstractProtocol.class.getDeclaredField("serverMap");
         serverMapField.setAccessible(true);
         Map<String, ProtocolServer> serverMap = (Map<String, ProtocolServer>) serverMapField.get(pocketMQProtocol);
         Assert.assertTrue(serverMap.isEmpty());

         RocketMQProtocolServer rocketMQProtocolServer = Whitebox.invokeMethod(pocketMQProtocol, "openServer", url, "");
         Assert.assertEquals(1, serverMap.size());
         PowerMockito.verifyPrivate(pocketMQProtocol).invoke("createServer", Mockito.any(URL.class), Mockito.anyString(), Mockito.anyString());

         RocketMQProtocolServer newRocketMQProtocolServer = Whitebox.invokeMethod(pocketMQProtocol, "openServer", url, "");
         Assert.assertEquals(rocketMQProtocolServer, newRocketMQProtocolServer);
     }

     @Test(expected = RuntimeException.class)
     public void createMessageSelectorTest() throws Exception {
         URL url = URLBuilder.from(this.url).addParameter(CommonConstants.GROUP_KEY, "test").build();
         MessageSelector messageSelector = Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", url);
         Assert.assertEquals("group=test", messageSelector.getExpression());

         url = URLBuilder.from(url).addParameter(CommonConstants.VERSION_KEY, "1.0.0").build();
         messageSelector = Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", url);
         Assert.assertEquals("group=test and version=1.0.0", messageSelector.getExpression());

         url = URLBuilder.from(this.url).addParameter(CommonConstants.VERSION_KEY, "1.0.0").build();
         messageSelector = Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", url);
         Assert.assertEquals("version=1.0.0", messageSelector.getExpression());

         Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", this.url);

     }

     @Test(expected = RpcException.class)
     public void exportTest() throws MQClientException {
         Invoker<Object> invoker = PowerMockito.mock(Invoker.class);
         PowerMockito.when(invoker.getUrl()).thenReturn(this.url);
         PowerMockito.when(rocketMQExporter.getKey()).thenReturn("test");

         DefaultMQPushConsumer consumer = PowerMockito.mock(DefaultMQPushConsumer.class);
         PowerMockito.when(this.rocketMQProtocolServer.getDefaultMQPushConsumer()).thenReturn(consumer);
         pocketMQProtocol.export(invoker);
         Mockito.verify(consumer, Mockito.atLeastOnce()).subscribe(Mockito.anyString(), Mockito.anyString());

         url = URLBuilder.from(url).addParameter("groupModel", "select").addParameter(CommonConstants.GROUP_KEY, "111").build();
         PowerMockito.when(invoker.getUrl()).thenReturn(this.url);
         pocketMQProtocol.export(invoker);
         Mockito.verify(consumer, Mockito.atLeastOnce()).subscribe(Mockito.anyString(), Mockito.any(MessageSelector.class));

         PowerMockito.doThrow(new RpcException()).when(consumer).subscribe(Mockito.anyString(), Mockito.any(MessageSelector.class));
         pocketMQProtocol.export(invoker);

     }

     @Test(expected = RpcException.class)
     public void exportExceptionTest() throws Exception {
         PowerMockito.when(pocketMQProtocol, "createServer", Mockito.any(URL.class), Mockito.anyString(), Mockito.anyString())
             .thenThrow(RpcException.class);
         Invoker<Object> invoker = PowerMockito.mock(Invoker.class);
         pocketMQProtocol.export(invoker);
     }


     @Test(expected = RpcException.class)
     public void protocolBindingReferTest() throws Exception {
         RocketMQInvoker<Object> invoker = PowerMockito.mock(RocketMQInvoker.class);
         PowerMockito.whenNew(RocketMQInvoker.class).withAnyArguments().thenReturn(invoker);

         PowerMockito.when(this.rocketMQProtocolServer.getDefaultMQProducer()).thenReturn(defaultMQProducer);

         PowerMockito.doReturn(this.rocketMQProtocolServer).when(pocketMQProtocol, "openServer", this.url, CommonConstants.CONSUMER);
         Invoker<Object> newInvoker = pocketMQProtocol.protocolBindingRefer(Object.class, this.url);
         PowerMockito.verifyPrivate(pocketMQProtocol).invoke("openServer", Mockito.any(URL.class), Mockito.anyString());
         Assert.assertEquals(newInvoker, invoker);


         PowerMockito.when(pocketMQProtocol, "openServer", this.url, CommonConstants.CONSUMER).thenThrow(RuntimeException.class);
         pocketMQProtocol.protocolBindingRefer(Object.class, this.url);

     }

     @Test
     public void consumeMessageTest() throws Exception {
         messageExt.putUserProperty(RocketMQProtocolConstant.SEND_ADDRESS, "127.0.0.1");
         ExecutorService executorService = PowerMockito.mock(ExecutorService.class);
         PowerMockito.when(rocketMQProtocolServer.getExecutorService()).thenReturn(executorService);
         PowerMockito.doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocation) throws Throwable {
                 Runnable runnable = invocation.getArgument(0);
                 runnable.run();
                 return null;
             }
         }).when(executorService).submit(Mockito.any(Runnable.class));
         messageListenerConcurrently.consumeMessage(this.msgs, null);

         Mockito.verify(executorService, Mockito.atLeastOnce()).submit(Mockito.any(Runnable.class));
         PowerMockito.verifyPrivate(messageListenerConcurrently).invoke("execute",Mockito.any(MessageExt.class));

     }

     @Test
     public void executeTest() throws Exception {
         messageExt.putUserProperty(RocketMQProtocolConstant.SEND_ADDRESS, "127.0.0.1");

         PowerMockito.when(messageListenerConcurrently, "invoke", Mockito.any(), Mockito.any(), Mockito.any()).thenReturn(new Response())
             .thenReturn(null);

         Whitebox.invokeMethod(messageListenerConcurrently, "execute", new Class[] {MessageExt.class},
             this.messageExt);

         Whitebox.invokeMethod(messageListenerConcurrently, "execute", new Class[] {MessageExt.class},
             this.messageExt);
     }


     @Test
     public void invokeTest() throws Exception {

         Channel channel = PowerMockito.mock(Channel.class);

         Response response = Whitebox.invokeMethod(messageListenerConcurrently, "invoke", new Class[] {MessageExt.class, Channel.class, URL.class},
             this.messageExt, channel, url);
         Assert.assertEquals(response.getStatus(), Response.BAD_REQUEST);

         Request request = PowerMockito.mock(Request.class);
         PowerMockito.when(codec.decode(Mockito.any(), Mockito.any())).thenReturn(request);
         Invocation inv = PowerMockito.mock(Invocation.class);
         PowerMockito.when(request.getData()).thenReturn(inv);

         Field exporterMapField = AbstractProtocol.class.getDeclaredField("exporterMap");
         exporterMapField.setAccessible(true);
         Map<String, Exporter<?>> exporterMap = (Map<String, Exporter<?>>) exporterMapField.get(pocketMQProtocol);
         Exporter<Object> exporter = PowerMockito.mock(Exporter.class);
         exporterMap.put("test", exporter);

         Invoker<Object> invoker = PowerMockito.mock(Invoker.class);
         PowerMockito.when(exporter.getInvoker()).thenReturn(invoker);

         Result result = PowerMockito.mock(Result.class);
         PowerMockito.when(invoker.invoke(Mockito.any(Invocation.class))).thenReturn(result);
         MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_CLUSTER, "");
         response = Whitebox.invokeMethod(messageListenerConcurrently, "invoke", new Class[] {MessageExt.class, Channel.class, URL.class},
             this.messageExt, channel, url);
         Assert.assertEquals(response.getResult(), result);

         messageExt.putUserProperty(CommonConstants.TIMEOUT_KEY, (System.currentTimeMillis() - 100000) + "");
         response = Whitebox.invokeMethod(messageListenerConcurrently, "invoke", new Class[] {MessageExt.class, Channel.class, URL.class},
             this.messageExt, channel, url);
         Assert.assertNull(response);
     }

     @Test
     public void createChannelBufferTest() throws Exception {
         this.response = new Response();
         Channel channel = PowerMockito.mock(Channel.class);

         ChannelBuffer channelBuffer =
             Whitebox.invokeMethod(messageListenerConcurrently, "createChannelBuffer", new Class[] {Channel.class, Response.class, URL.class},
                 channel, response, url);
         Assert.assertNotNull(channelBuffer);
         Mockito.verify(codec, Mockito.atLeastOnce()).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));

         Mockito.doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocation)  {
                 Response newResponse = invocation.getArgument(2);
                 Assert.assertEquals(newResponse, response);
                 throw new RuntimeException();

             }
         }).doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocation)  {
                 Response newResponse = invocation.getArgument(2);
                 Assert.assertEquals(newResponse.getStatus(), Response.BAD_REQUEST);
                 return null;
             }
         }).when(codec).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Response.class));

         channelBuffer =
             Whitebox.invokeMethod(messageListenerConcurrently, "createChannelBuffer", new Class[] {Channel.class, Response.class, URL.class},
                 channel, response, this.url);
         Assert.assertNotNull(channelBuffer);
         Mockito.verify(codec, Mockito.atLeast(2)).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));

         Mockito.reset(codec);
         PowerMockito.doThrow(new RuntimeException()).doThrow(new IOException()).when(codec)
             .encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));
         channelBuffer =
             Whitebox.invokeMethod(messageListenerConcurrently, "createChannelBuffer", new Class[] {Channel.class, Response.class, URL.class},
                 channel, response, this.url);
         Assert.assertNull(channelBuffer);
         Mockito.verify(codec, Mockito.atLeast(2)).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));
     }

     @Test
     public void sendMessageTest() throws Exception {
         MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_CLUSTER, "");
         ChannelBuffer buffer = PowerMockito.mock(ChannelBuffer.class);

         PowerMockito.doAnswer(new Answer<Object>() {
             @Override
             public Object answer(InvocationOnMock invocation)  {
                 long time = invocation.getArgument(1);
                 Assert.assertEquals(time, 3000L);
                 return null;
             }
         }).when(this.defaultMQProducer).send(Mockito.any(Message.class), Mockito.anyLong());
         boolean value = Whitebox
             .invokeMethod(messageListenerConcurrently, "sendMessage", new Class[] {MessageExt.class, ChannelBuffer.class, URL.class, String.class},
                 this.messageExt, buffer, this.url, "url");
         Assert.assertTrue(value);
         Mockito.verify(defaultMQProducer, Mockito.atLeastOnce()).send(Mockito.any(Message.class), Mockito.anyLong());

         PowerMockito.when(this.defaultMQProducer.send(Mockito.any(Message.class), Mockito.anyLong())).thenThrow(new RuntimeException());
         value = Whitebox
             .invokeMethod(messageListenerConcurrently, "sendMessage", new Class[] {MessageExt.class, ChannelBuffer.class, URL.class, String.class},
                 this.messageExt, buffer, this.url, "url");
         Assert.assertFalse(value);
     }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.dubbo.rpc.rocketmq;

	import org.apache.dubbo.common.URL;
	import org.apache.dubbo.common.URLBuilder;
	import org.apache.dubbo.common.constants.CommonConstants;
	import org.apache.dubbo.remoting.Channel;
	import org.apache.dubbo.remoting.buffer.ChannelBuffer;
	import org.apache.dubbo.remoting.exchange.Request;
	import org.apache.dubbo.remoting.exchange.Response;
	import org.apache.dubbo.rpc.Exporter;
	import org.apache.dubbo.rpc.Invocation;
	import org.apache.dubbo.rpc.Invoker;
	import org.apache.dubbo.rpc.ProtocolServer;
	import org.apache.dubbo.rpc.Result;
	import org.apache.dubbo.rpc.RpcException;
	import org.apache.dubbo.rpc.protocol.AbstractProtocol;
	import org.apache.dubbo.rpc.rocketmq.codec.RocketMQCountCodec;
	import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
	import org.apache.rocketmq.client.consumer.MessageSelector;
	import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
	import org.apache.rocketmq.client.exception.MQClientException;
	import org.apache.rocketmq.client.producer.DefaultMQProducer;
	import org.apache.rocketmq.common.message.Message;
	import org.apache.rocketmq.common.message.MessageAccessor;
	import org.apache.rocketmq.common.message.MessageConst;
	import org.apache.rocketmq.common.message.MessageExt;

	import java.io.IOException;
	import java.lang.reflect.Field;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.Map;
	import java.util.concurrent.ExecutorService;


	import org.junit.Assert;
	import org.junit.Before;
	import org.junit.Test;
	import org.junit.runner.RunWith;
	import org.mockito.Mockito;
	import org.mockito.invocation.InvocationOnMock;
	import org.mockito.stubbing.Answer;
	import org.powermock.api.mockito.PowerMockito;
	import org.powermock.core.classloader.annotations.PrepareForTest;
	import org.powermock.modules.junit4.PowerMockRunner;
	import org.powermock.reflect.Whitebox;

	@RunWith(PowerMockRunner.class)
	@PrepareForTest(value = {RocketMQProtocol.class})
	public class RocketMQProtocolTest {

	private final RocketMQProtocol pocketMQProtocol = PowerMockito.spy(new RocketMQProtocol());

	private final DefaultMQProducer defaultMQProducer = PowerMockito.mock(DefaultMQProducer.class);

	private final RocketMQProtocolServer rocketMQProtocolServer = PowerMockito.mock(RocketMQProtocolServer.class);

	private final RocketMQExporter<Object> rocketMQExporter = PowerMockito.mock(RocketMQExporter.class);

	private final RocketMQCountCodec codec = PowerMockito.mock(RocketMQCountCodec.class);
	private final List<MessageExt> msgs = new ArrayList<>();
	private final MessageExt messageExt = new MessageExt();
	private MessageListenerConcurrently messageListenerConcurrently;
	private Response response;

	private URL url;


	@Before
	public void before() throws Exception {
	String urlString =
	"nameservice://localhost:9876/org.apache.dubbo.registry.RegistryService?application=rocketmq-provider&dubbo=2.0.2&interface=org.apache.dubbo.registry.RegistryService&pid=8990&release=3.0.7&route=false";
	url = URLBuilder.valueOf(urlString).addParameter("groupModel", "topic").addParameter(CommonConstants.ENABLE_TIMEOUT_COUNTDOWN_KEY, false);

	PowerMockito.whenNew(RocketMQProtocolServer.class).withAnyArguments().thenReturn(rocketMQProtocolServer);
	PowerMockito.when(rocketMQProtocolServer.getDefaultMQProducer()).thenReturn(defaultMQProducer);
	PowerMockito.whenNew(RocketMQExporter.class).withAnyArguments().thenReturn(rocketMQExporter);

	PowerMockito.whenNew(RocketMQCountCodec.class).withAnyArguments().thenReturn(codec);
	PowerMockito.doAnswer(new Answer() {
	@Override
	public Object answer(InvocationOnMock invocation) {
	messageListenerConcurrently = invocation.getArgument(0);
	messageListenerConcurrently = PowerMockito.spy(messageListenerConcurrently);
	return null;
	}
	}).when(rocketMQProtocolServer).setMessageListenerConcurrently(Mockito.any(MessageListenerConcurrently.class));
	PowerMockito.when(pocketMQProtocol, "createServer", Mockito.any(URL.class), Mockito.anyString(), Mockito.anyString()).thenCallRealMethod();

	messageExt.setTopic("test");
	messageExt.putUserProperty(CommonConstants.TIMEOUT_KEY, (System.currentTimeMillis() + 100000) + "");
	messageExt.putUserProperty(RocketMQProtocolConstant.URL_STRING, urlString);
	messageExt.setBody(new byte[1024]);

	msgs.add(messageExt);

	}

	@Test
	public void createServerTest() throws Exception {
	PowerMockito.when(pocketMQProtocol, "createServer", url, "", "").thenCallRealMethod();

	Mockito.verify(rocketMQProtocolServer, Mockito.atLeastOnce()).setModel(Mockito.any(String.class));
	Mockito.verify(rocketMQProtocolServer, Mockito.atLeastOnce()).setMessageListenerConcurrently(Mockito.any(MessageListenerConcurrently.class));
	Mockito.verify(rocketMQProtocolServer, Mockito.atLeastOnce()).reset(url);

	}

	@Test
	public void openServerTest() throws Exception {
	Field serverMapField = AbstractProtocol.class.getDeclaredField("serverMap");
	serverMapField.setAccessible(true);
	Map<String, ProtocolServer> serverMap = (Map<String, ProtocolServer>) serverMapField.get(pocketMQProtocol);
	Assert.assertTrue(serverMap.isEmpty());

	RocketMQProtocolServer rocketMQProtocolServer = Whitebox.invokeMethod(pocketMQProtocol, "openServer", url, "");
	Assert.assertEquals(1, serverMap.size());
	PowerMockito.verifyPrivate(pocketMQProtocol).invoke("createServer", Mockito.any(URL.class), Mockito.anyString(), Mockito.anyString());

	RocketMQProtocolServer newRocketMQProtocolServer = Whitebox.invokeMethod(pocketMQProtocol, "openServer", url, "");
	Assert.assertEquals(rocketMQProtocolServer, newRocketMQProtocolServer);
	}

	@Test(expected = RuntimeException.class)
	public void createMessageSelectorTest() throws Exception {
	URL url = URLBuilder.from(this.url).addParameter(CommonConstants.GROUP_KEY, "test").build();
	MessageSelector messageSelector = Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", url);
	Assert.assertEquals("group=test", messageSelector.getExpression());

	url = URLBuilder.from(url).addParameter(CommonConstants.VERSION_KEY, "1.0.0").build();
	messageSelector = Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", url);
	Assert.assertEquals("group=test and version=1.0.0", messageSelector.getExpression());

	url = URLBuilder.from(this.url).addParameter(CommonConstants.VERSION_KEY, "1.0.0").build();
	messageSelector = Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", url);
	Assert.assertEquals("version=1.0.0", messageSelector.getExpression());

	Whitebox.invokeMethod(pocketMQProtocol, "createMessageSelector", this.url);

	}

	@Test(expected = RpcException.class)
	public void exportTest() throws MQClientException {
	Invoker<Object> invoker = PowerMockito.mock(Invoker.class);
	PowerMockito.when(invoker.getUrl()).thenReturn(this.url);
	PowerMockito.when(rocketMQExporter.getKey()).thenReturn("test");

	DefaultMQPushConsumer consumer = PowerMockito.mock(DefaultMQPushConsumer.class);
	PowerMockito.when(this.rocketMQProtocolServer.getDefaultMQPushConsumer()).thenReturn(consumer);
	pocketMQProtocol.export(invoker);
	Mockito.verify(consumer, Mockito.atLeastOnce()).subscribe(Mockito.anyString(), Mockito.anyString());

	url = URLBuilder.from(url).addParameter("groupModel", "select").addParameter(CommonConstants.GROUP_KEY, "111").build();
	PowerMockito.when(invoker.getUrl()).thenReturn(this.url);
	pocketMQProtocol.export(invoker);
	Mockito.verify(consumer, Mockito.atLeastOnce()).subscribe(Mockito.anyString(), Mockito.any(MessageSelector.class));

	PowerMockito.doThrow(new RpcException()).when(consumer).subscribe(Mockito.anyString(), Mockito.any(MessageSelector.class));
	pocketMQProtocol.export(invoker);

	}

	@Test(expected = RpcException.class)
	public void exportExceptionTest() throws Exception {
	PowerMockito.when(pocketMQProtocol, "createServer", Mockito.any(URL.class), Mockito.anyString(), Mockito.anyString())
	.thenThrow(RpcException.class);
	Invoker<Object> invoker = PowerMockito.mock(Invoker.class);
	pocketMQProtocol.export(invoker);
	}


	@Test(expected = RpcException.class)
	public void protocolBindingReferTest() throws Exception {
	RocketMQInvoker<Object> invoker = PowerMockito.mock(RocketMQInvoker.class);
	PowerMockito.whenNew(RocketMQInvoker.class).withAnyArguments().thenReturn(invoker);

	PowerMockito.when(this.rocketMQProtocolServer.getDefaultMQProducer()).thenReturn(defaultMQProducer);

	PowerMockito.doReturn(this.rocketMQProtocolServer).when(pocketMQProtocol, "openServer", this.url, CommonConstants.CONSUMER);
	Invoker<Object> newInvoker = pocketMQProtocol.protocolBindingRefer(Object.class, this.url);
	PowerMockito.verifyPrivate(pocketMQProtocol).invoke("openServer", Mockito.any(URL.class), Mockito.anyString());
	Assert.assertEquals(newInvoker, invoker);


	PowerMockito.when(pocketMQProtocol, "openServer", this.url, CommonConstants.CONSUMER).thenThrow(RuntimeException.class);
	pocketMQProtocol.protocolBindingRefer(Object.class, this.url);

	}

	@Test
	public void consumeMessageTest() throws Exception {
	messageExt.putUserProperty(RocketMQProtocolConstant.SEND_ADDRESS, "127.0.0.1");
	ExecutorService executorService = PowerMockito.mock(ExecutorService.class);
	PowerMockito.when(rocketMQProtocolServer.getExecutorService()).thenReturn(executorService);
	PowerMockito.doAnswer(new Answer<Object>() {
	@Override
	public Object answer(InvocationOnMock invocation) throws Throwable {
	Runnable runnable = invocation.getArgument(0);
	runnable.run();
	return null;
	}
	}).when(executorService).submit(Mockito.any(Runnable.class));
	messageListenerConcurrently.consumeMessage(this.msgs, null);

	Mockito.verify(executorService, Mockito.atLeastOnce()).submit(Mockito.any(Runnable.class));
	PowerMockito.verifyPrivate(messageListenerConcurrently).invoke("execute",Mockito.any(MessageExt.class));

	}

	@Test
	public void executeTest() throws Exception {
	messageExt.putUserProperty(RocketMQProtocolConstant.SEND_ADDRESS, "127.0.0.1");

	PowerMockito.when(messageListenerConcurrently, "invoke", Mockito.any(), Mockito.any(), Mockito.any()).thenReturn(new Response())
	.thenReturn(null);

	Whitebox.invokeMethod(messageListenerConcurrently, "execute", new Class[] {MessageExt.class},
	this.messageExt);

	Whitebox.invokeMethod(messageListenerConcurrently, "execute", new Class[] {MessageExt.class},
	this.messageExt);
	}


	@Test
	public void invokeTest() throws Exception {

	Channel channel = PowerMockito.mock(Channel.class);

	Response response = Whitebox.invokeMethod(messageListenerConcurrently, "invoke", new Class[] {MessageExt.class, Channel.class, URL.class},
	this.messageExt, channel, url);
	Assert.assertEquals(response.getStatus(), Response.BAD_REQUEST);

	Request request = PowerMockito.mock(Request.class);
	PowerMockito.when(codec.decode(Mockito.any(), Mockito.any())).thenReturn(request);
	Invocation inv = PowerMockito.mock(Invocation.class);
	PowerMockito.when(request.getData()).thenReturn(inv);

	Field exporterMapField = AbstractProtocol.class.getDeclaredField("exporterMap");
	exporterMapField.setAccessible(true);
	Map<String, Exporter<?>> exporterMap = (Map<String, Exporter<?>>) exporterMapField.get(pocketMQProtocol);
	Exporter<Object> exporter = PowerMockito.mock(Exporter.class);
	exporterMap.put("test", exporter);

	Invoker<Object> invoker = PowerMockito.mock(Invoker.class);
	PowerMockito.when(exporter.getInvoker()).thenReturn(invoker);

	Result result = PowerMockito.mock(Result.class);
	PowerMockito.when(invoker.invoke(Mockito.any(Invocation.class))).thenReturn(result);
	MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_CLUSTER, "");
	response = Whitebox.invokeMethod(messageListenerConcurrently, "invoke", new Class[] {MessageExt.class, Channel.class, URL.class},
	this.messageExt, channel, url);
	Assert.assertEquals(response.getResult(), result);

	messageExt.putUserProperty(CommonConstants.TIMEOUT_KEY, (System.currentTimeMillis() - 100000) + "");
	response = Whitebox.invokeMethod(messageListenerConcurrently, "invoke", new Class[] {MessageExt.class, Channel.class, URL.class},
	this.messageExt, channel, url);
	Assert.assertNull(response);
	}

	@Test
	public void createChannelBufferTest() throws Exception {
	this.response = new Response();
	Channel channel = PowerMockito.mock(Channel.class);

	ChannelBuffer channelBuffer =
	Whitebox.invokeMethod(messageListenerConcurrently, "createChannelBuffer", new Class[] {Channel.class, Response.class, URL.class},
	channel, response, url);
	Assert.assertNotNull(channelBuffer);
	Mockito.verify(codec, Mockito.atLeastOnce()).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));

	Mockito.doAnswer(new Answer<Object>() {
	@Override
	public Object answer(InvocationOnMock invocation) {
	Response newResponse = invocation.getArgument(2);
	Assert.assertEquals(newResponse, response);
	throw new RuntimeException();

	}
	}).doAnswer(new Answer<Object>() {
	@Override
	public Object answer(InvocationOnMock invocation) {
	Response newResponse = invocation.getArgument(2);
	Assert.assertEquals(newResponse.getStatus(), Response.BAD_REQUEST);
	return null;
	}
	}).when(codec).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Response.class));

	channelBuffer =
	Whitebox.invokeMethod(messageListenerConcurrently, "createChannelBuffer", new Class[] {Channel.class, Response.class, URL.class},
	channel, response, this.url);
	Assert.assertNotNull(channelBuffer);
	Mockito.verify(codec, Mockito.atLeast(2)).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));

	Mockito.reset(codec);
	PowerMockito.doThrow(new RuntimeException()).doThrow(new IOException()).when(codec)
	.encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));
	channelBuffer =
	Whitebox.invokeMethod(messageListenerConcurrently, "createChannelBuffer", new Class[] {Channel.class, Response.class, URL.class},
	channel, response, this.url);
	Assert.assertNull(channelBuffer);
	Mockito.verify(codec, Mockito.atLeast(2)).encode(Mockito.any(Channel.class), Mockito.any(ChannelBuffer.class), Mockito.any(Object.class));
	}

	@Test
	public void sendMessageTest() throws Exception {
	MessageAccessor.putProperty(messageExt, MessageConst.PROPERTY_CLUSTER, "");
	ChannelBuffer buffer = PowerMockito.mock(ChannelBuffer.class);

	PowerMockito.doAnswer(new Answer<Object>() {
	@Override
	public Object answer(InvocationOnMock invocation) {
	long time = invocation.getArgument(1);
	Assert.assertEquals(time, 3000L);
	return null;
	}
	}).when(this.defaultMQProducer).send(Mockito.any(Message.class), Mockito.anyLong());
	boolean value = Whitebox
	.invokeMethod(messageListenerConcurrently, "sendMessage", new Class[] {MessageExt.class, ChannelBuffer.class, URL.class, String.class},
	this.messageExt, buffer, this.url, "url");
	Assert.assertTrue(value);
	Mockito.verify(defaultMQProducer, Mockito.atLeastOnce()).send(Mockito.any(Message.class), Mockito.anyLong());

	PowerMockito.when(this.defaultMQProducer.send(Mockito.any(Message.class), Mockito.anyLong())).thenThrow(new RuntimeException());
	value = Whitebox
	.invokeMethod(messageListenerConcurrently, "sendMessage", new Class[] {MessageExt.class, ChannelBuffer.class, URL.class, String.class},
	this.messageExt, buffer, this.url, "url");
	Assert.assertFalse(value);
	}

	}