diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/WebCluster.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/WebCluster.java index 31a2fd226..eda34a4b1 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/WebCluster.java +++ b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/WebCluster.java @@ -22,13 +22,13 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.navercorp.pinpoint.rpc.PinpointSocketException; import com.navercorp.pinpoint.rpc.client.MessageListener; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; +import com.navercorp.pinpoint.rpc.client.PinpointClient; import com.navercorp.pinpoint.rpc.stream.DisabledServerStreamChannelMessageListener; import com.navercorp.pinpoint.rpc.stream.ServerStreamChannelMessageListener; @@ -38,24 +38,24 @@ import com.navercorp.pinpoint.rpc.stream.ServerStreamChannelMessageListener; public class WebCluster implements Cluster { private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private final PinpointSocketFactory factory; + private final PinpointClientFactory clientFactory; - private final Map clusterRepository = new HashMap(); + private final Map clusterRepository = new HashMap(); public WebCluster(String id, MessageListener messageListener) { this(id, messageListener, DisabledServerStreamChannelMessageListener.INSTANCE); } public WebCluster(String id, MessageListener messageListener, ServerStreamChannelMessageListener serverStreamChannelMessageListener) { - this.factory = new PinpointSocketFactory(); - this.factory.setTimeoutMillis(1000 * 5); - this.factory.setMessageListener(messageListener); - this.factory.setServerStreamChannelMessageListener(serverStreamChannelMessageListener); + this.clientFactory = new PinpointClientFactory(); + this.clientFactory.setTimeoutMillis(1000 * 5); + this.clientFactory.setMessageListener(messageListener); + this.clientFactory.setServerStreamChannelMessageListener(serverStreamChannelMessageListener); Map properties = new HashMap(); properties.put("id", id); - factory.setProperties(properties); + clientFactory.setProperties(properties); } // Not safe for use by multiple threads. @@ -67,8 +67,8 @@ public class WebCluster implements Cluster { return; } - PinpointSocket socket = createPinpointSocket(address); - clusterRepository.put(address, socket); + PinpointClient client = createPinpointClient(address); + clusterRepository.put(address, client); logger.info("localhost -> {} connect completed.", address); } @@ -77,7 +77,7 @@ public class WebCluster implements Cluster { public void disconnectPoint(InetSocketAddress address) { logger.info("localhost -> {} disconnect started.", address); - PinpointSocket socket = clusterRepository.remove(address); + PinpointClient socket = clusterRepository.remove(address); if (socket != null) { socket.close(); logger.info("localhost -> {} disconnect completed.", address); @@ -86,24 +86,24 @@ public class WebCluster implements Cluster { } } - private PinpointSocket createPinpointSocket(InetSocketAddress address) { + private PinpointClient createPinpointClient(InetSocketAddress address) { String host = address.getHostName(); int port = address.getPort(); - PinpointSocket socket = null; + PinpointClient client = null; for (int i = 0; i < 3; i++) { try { - socket = factory.connect(host, port); + client = clientFactory.connect(host, port); logger.info("tcp connect success:{}/{}", host, port); - return socket; + return client; } catch (PinpointSocketException e) { logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i); } } logger.warn("change background tcp connect mode {}/{} ", host, port); - socket = factory.scheduledConnect(host, port); + client = clientFactory.scheduledConnect(host, port); - return socket; + return client; } public List getWebClusterList() { @@ -111,14 +111,14 @@ public class WebCluster implements Cluster { } public void close() { - for (PinpointSocket socket : clusterRepository.values()) { - if (socket != null) { - socket.close(); + for (PinpointClient client : clusterRepository.values()) { + if (client != null) { + client.close(); } } - if (factory != null) { - factory.release(); + if (clientFactory != null) { + clientFactory.release(); } } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java index fb896a636..a81c786b2 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java @@ -16,13 +16,13 @@ package com.navercorp.pinpoint.profiler; -import java.lang.instrument.ClassFileTransformer; import java.lang.instrument.Instrumentation; import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; +import com.navercorp.pinpoint.rpc.client.PinpointClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -65,8 +65,7 @@ import com.navercorp.pinpoint.profiler.util.ApplicationServerTypeResolver; import com.navercorp.pinpoint.profiler.util.RuntimeMXBeanUtils; import com.navercorp.pinpoint.rpc.ClassPreLoader; import com.navercorp.pinpoint.rpc.PinpointSocketException; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; /** * @author emeroad @@ -88,8 +87,8 @@ public class DefaultAgent implements Agent { private final TraceContext traceContext; - private PinpointSocketFactory factory; - private PinpointSocket socket; + private PinpointClientFactory clientFactory; + private PinpointClient client; private final EnhancedDataSender tcpDataSender; private final DataSender statDataSender; @@ -304,48 +303,48 @@ public class DefaultAgent implements Agent { return serverMetaDataHolder; } - protected PinpointSocketFactory createPinpointSocketFactory(CommandDispatcher commandDispatcher) { - PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setTimeoutMillis(1000 * 5); + protected PinpointClientFactory createPinpointClientFactory(CommandDispatcher commandDispatcher) { + PinpointClientFactory pinpointClientFactory = new PinpointClientFactory(); + pinpointClientFactory.setTimeoutMillis(1000 * 5); Map properties = this.agentInformation.toMap(); boolean isSupportServerMode = this.profilerConfig.isTcpDataSenderCommandAcceptEnable(); if (isSupportServerMode) { - pinpointSocketFactory.setMessageListener(commandDispatcher); - pinpointSocketFactory.setServerStreamChannelMessageListener(commandDispatcher); + pinpointClientFactory.setMessageListener(commandDispatcher); + pinpointClientFactory.setServerStreamChannelMessageListener(commandDispatcher); properties.put(AgentHandshakePropertyType.SUPPORT_SERVER.getName(), true); } else { properties.put(AgentHandshakePropertyType.SUPPORT_SERVER.getName(), false); } - pinpointSocketFactory.setProperties(properties); - return pinpointSocketFactory; + pinpointClientFactory.setProperties(properties); + return pinpointClientFactory; } - protected PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) { - PinpointSocket socket = null; + protected PinpointClient createPinpointClient(String host, int port, PinpointClientFactory factory) { + PinpointClient client = null; for (int i = 0; i < 3; i++) { try { - socket = factory.connect(host, port); + client = factory.connect(host, port); logger.info("tcp connect success:{}/{}", host, port); - return socket; + return client; } catch (PinpointSocketException e) { logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i); } } logger.warn("change background tcp connect mode {}/{} ", host, port); - socket = factory.scheduledConnect(host, port); + client = factory.scheduledConnect(host, port); - return socket; + return client; } protected EnhancedDataSender createTcpDataSender(CommandDispatcher commandDispatcher) { - this.factory = createPinpointSocketFactory(commandDispatcher); - this.socket = createPinpointSocket(this.profilerConfig.getCollectorTcpServerIp(), this.profilerConfig.getCollectorTcpServerPort(), factory); - return new TcpDataSender(socket); + this.clientFactory = createPinpointClientFactory(commandDispatcher); + this.client = createPinpointClient(this.profilerConfig.getCollectorTcpServerIp(), this.profilerConfig.getCollectorTcpServerPort(), clientFactory); + return new TcpDataSender(client); } protected DataSender createUdpStatDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize) { @@ -424,11 +423,11 @@ public class DefaultAgent implements Agent { if (this.tcpDataSender != null) { this.tcpDataSender.stop(); } - if (this.socket != null) { - this.socket.close(); + if (this.client != null) { + this.client.close(); } - if (this.factory != null) { - this.factory.release(); + if (this.clientFactory != null) { + this.clientFactory.release(); } } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EnhancedDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EnhancedDataSender.java index abf64724e..0d4511fc7 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EnhancedDataSender.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EnhancedDataSender.java @@ -18,7 +18,7 @@ package com.navercorp.pinpoint.profiler.sender; import com.navercorp.pinpoint.rpc.FutureListener; import com.navercorp.pinpoint.rpc.ResponseMessage; -import com.navercorp.pinpoint.rpc.client.PinpointSocketReconnectEventListener; +import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener; import org.apache.thrift.TBase; @@ -31,7 +31,7 @@ public interface EnhancedDataSender extends DataSender { boolean request(TBase data, int retry); boolean request(TBase data, FutureListener listener); - boolean addReconnectEventListener(PinpointSocketReconnectEventListener eventListener); - boolean removeReconnectEventListener(PinpointSocketReconnectEventListener eventListener); + boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener); + boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener); } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/LoggingDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/LoggingDataSender.java index 2a946b5a5..d4c004833 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/LoggingDataSender.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/LoggingDataSender.java @@ -18,7 +18,7 @@ package com.navercorp.pinpoint.profiler.sender; import com.navercorp.pinpoint.rpc.FutureListener; import com.navercorp.pinpoint.rpc.ResponseMessage; -import com.navercorp.pinpoint.rpc.client.PinpointSocketReconnectEventListener; +import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener; import org.apache.thrift.TBase; import org.slf4j.Logger; @@ -67,13 +67,13 @@ public class LoggingDataSender implements EnhancedDataSender { } @Override - public boolean addReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) { logger.info("addReconnectEventListener eventListener:{}", eventListener); return false; } @Override - public boolean removeReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) { logger.info("removeReconnectEventListener eventListener:{}", eventListener); return false; } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java index 8125817b5..642ace215 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java @@ -33,8 +33,8 @@ import org.slf4j.LoggerFactory; import com.navercorp.pinpoint.rpc.Future; import com.navercorp.pinpoint.rpc.FutureListener; import com.navercorp.pinpoint.rpc.ResponseMessage; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketReconnectEventListener; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener; import com.navercorp.pinpoint.rpc.util.TimerFactory; import com.navercorp.pinpoint.thrift.dto.TResult; import com.navercorp.pinpoint.thrift.io.HeaderTBaseDeserializer; @@ -55,7 +55,7 @@ public class TcpDataSender extends AbstractDataSender implements EnhancedDataSen ChannelBuffers.buffer(2); } - private final PinpointSocket socket; + private final PinpointClient client; private final Timer timer; private final AtomicBoolean fireState = new AtomicBoolean(false); @@ -69,8 +69,8 @@ public class TcpDataSender extends AbstractDataSender implements EnhancedDataSen private AsyncQueueingExecutor executor; - public TcpDataSender(PinpointSocket socket) { - this.socket = socket; + public TcpDataSender(PinpointClient client) { + this.client = client; this.timer = createTimer(); writeFailFutureListener = new WriteFailFutureListener(logger, "io write fail.", "host", -1); this.executor = createAsyncQueueingExecutor(1024 * 5, "Pinpoint-TcpDataExecutor"); @@ -105,13 +105,13 @@ public class TcpDataSender extends AbstractDataSender implements EnhancedDataSen } @Override - public boolean addReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { - return this.socket.addPinpointSocketReconnectEventListener(eventListener); + public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) { + return this.client.addPinpointClientReconnectEventListener(eventListener); } @Override - public boolean removeReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { - return this.socket.removePinpointSocketReconnectEventListener(eventListener); + public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) { + return this.client.removePinpointClientReconnectEventListener(eventListener); } @Override @@ -159,7 +159,7 @@ public class TcpDataSender extends AbstractDataSender implements EnhancedDataSen } private void doSend(byte[] copy) { - Future write = this.socket.sendAsync(copy); + Future write = this.client.sendAsync(copy); write.setListener(writeFailFutureListener); } @@ -218,7 +218,7 @@ public class TcpDataSender extends AbstractDataSender implements EnhancedDataSen } private void doRequest(final byte[] requestPacket, FutureListener futureListener) { - final Future response = this.socket.request(requestPacket); + final Future response = this.client.request(requestPacket); response.setListener(futureListener); } @@ -237,9 +237,9 @@ public class TcpDataSender extends AbstractDataSender implements EnhancedDataSen @Override public boolean isNetworkAvailable() { - if (this.socket == null) { + if (this.client == null) { return false; } - return this.socket.isConnected(); + return this.client.isConnected(); } } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/tools/NetworkAvailabilityChecker.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/tools/NetworkAvailabilityChecker.java index 56c77f131..e1097517a 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/tools/NetworkAvailabilityChecker.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/tools/NetworkAvailabilityChecker.java @@ -18,6 +18,8 @@ package com.navercorp.pinpoint.profiler.tools; import java.util.Collections; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -26,8 +28,6 @@ import com.navercorp.pinpoint.profiler.sender.DataSender; import com.navercorp.pinpoint.profiler.sender.TcpDataSender; import com.navercorp.pinpoint.profiler.sender.UdpDataSender; import com.navercorp.pinpoint.rpc.PinpointSocketException; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; /** * @@ -50,8 +50,8 @@ public class NetworkAvailabilityChecker implements PinpointTools { DataSender udpSpanSender = null; DataSender tcpSender = null; - PinpointSocketFactory socketFactory = null; - PinpointSocket socket = null; + PinpointClientFactory clientFactory = null; + PinpointClient client = null; try { ProfilerConfig profilerConfig = ProfilerConfig.load(configPath); @@ -65,10 +65,10 @@ public class NetworkAvailabilityChecker implements PinpointTools { String collectorTcpIp = profilerConfig.getCollectorTcpServerIp(); int collectorTcpPort = profilerConfig.getCollectorTcpServerPort(); - socketFactory = createPinpointSocketFactory(); - socket = createPinpointSocket(collectorTcpIp, collectorTcpPort, socketFactory); + clientFactory = createPinpointClientFactory(); + client = createPinpointClient(collectorTcpIp, collectorTcpPort, clientFactory); - tcpSender = new TcpDataSender(socket); + tcpSender = new TcpDataSender(client); boolean udpSenderResult = udpStatSender.isNetworkAvailable(); boolean udpSpanSenderResult = udpSpanSender.isNetworkAvailable(); @@ -89,11 +89,11 @@ public class NetworkAvailabilityChecker implements PinpointTools { closeDataSender(tcpSender); System.out.println("END."); - if (socket != null) { - socket.close(); + if (client != null) { + client.close(); } - if (socketFactory != null) { - socketFactory.release(); + if (clientFactory != null) { + clientFactory.release(); } } } @@ -114,23 +114,22 @@ public class NetworkAvailabilityChecker implements PinpointTools { } } - private static PinpointSocketFactory createPinpointSocketFactory() { - PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setTimeoutMillis(1000 * 5); - pinpointSocketFactory.setProperties(Collections.emptyMap()); + private static PinpointClientFactory createPinpointClientFactory() { + PinpointClientFactory pinpointClientFactory = new PinpointClientFactory(); + pinpointClientFactory.setTimeoutMillis(1000 * 5); + pinpointClientFactory.setProperties(Collections.emptyMap()); - return pinpointSocketFactory; + return pinpointClientFactory; } - private static PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) { - + private static PinpointClient createPinpointClient(String host, int port, PinpointClientFactory factory) { RuntimeException lastException = null; for (int i = 0; i < 3; i++) { try { - PinpointSocket socket = factory.connect(host, port); + PinpointClient pinpointClient = factory.connect(host, port); LOGGER.info("tcp connect success:{}/{}", host, port); - return socket; + return pinpointClient; } catch (PinpointSocketException e) { LOGGER.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i); lastException = e; diff --git a/profiler/src/main/java/com/navercorp/pinpoint/test/TestTcpDataSender.java b/profiler/src/main/java/com/navercorp/pinpoint/test/TestTcpDataSender.java index 967838ada..f7837b0c4 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/test/TestTcpDataSender.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/test/TestTcpDataSender.java @@ -29,7 +29,7 @@ import org.apache.thrift.TBase; import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender; import com.navercorp.pinpoint.rpc.FutureListener; import com.navercorp.pinpoint.rpc.ResponseMessage; -import com.navercorp.pinpoint.rpc.client.PinpointSocketReconnectEventListener; +import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener; import com.navercorp.pinpoint.thrift.dto.TApiMetaData; import com.navercorp.pinpoint.thrift.dto.TSqlMetaData; import com.navercorp.pinpoint.thrift.dto.TStringMetaData; @@ -126,12 +126,12 @@ public class TestTcpDataSender implements EnhancedDataSender { } @Override - public boolean addReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) { return false; } @Override - public boolean removeReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) { return false; } diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java index d3b72a6a6..3290737c5 100644 --- a/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java +++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java @@ -40,6 +40,8 @@ import com.navercorp.pinpoint.common.trace.ServiceType; import com.navercorp.pinpoint.common.util.JvmUtils; import com.navercorp.pinpoint.common.util.SystemPropertyKey; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.apache.thrift.TException; import org.junit.Test; import org.slf4j.Logger; @@ -49,8 +51,6 @@ import com.navercorp.pinpoint.profiler.context.DefaultServerMetaData; import com.navercorp.pinpoint.profiler.context.DefaultServerMetaDataHolder; import com.navercorp.pinpoint.profiler.sender.TcpDataSender; import com.navercorp.pinpoint.rpc.PinpointSocketException; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseType; import com.navercorp.pinpoint.rpc.packet.PingPacket; @@ -80,17 +80,17 @@ public class AgentInfoSenderTest { PinpointServerAcceptor serverAcceptor = createServerAcceptor(serverListener); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory clientFactory = createPinpointClientFactory(); + PinpointClient pinpointClient = createPinpointClient(HOST, PORT, clientFactory); - TcpDataSender sender = new TcpDataSender(socket); + TcpDataSender sender = new TcpDataSender(pinpointClient); AgentInfoSender agentInfoSender = new AgentInfoSender(sender, agentInfoSendRetryIntervalMs, getAgentInfo()); try { agentInfoSender.start(); Thread.sleep(10000L); } finally { - closeAll(serverAcceptor, agentInfoSender, socket, socketFactory); + closeAll(serverAcceptor, agentInfoSender, pinpointClient, clientFactory); } assertEquals(1, requestCount.get()); assertEquals(1, successCount.get()); @@ -107,17 +107,17 @@ public class AgentInfoSenderTest { PinpointServerAcceptor serverAcceptor = createServerAcceptor(serverListener); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory socketFactory = createPinpointClientFactory(); + PinpointClient pinpointClient = createPinpointClient(HOST, PORT, socketFactory); - TcpDataSender dataSender = new TcpDataSender(socket); + TcpDataSender dataSender = new TcpDataSender(pinpointClient); AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo()); try { agentInfoSender.start(); Thread.sleep(agentInfoSendRetryIntervalMs * expectedTriesUntilSuccess); } finally { - closeAll(serverAcceptor, agentInfoSender, socket, socketFactory); + closeAll(serverAcceptor, agentInfoSender, pinpointClient, socketFactory); } assertEquals(expectedTriesUntilSuccess, requestCount.get()); assertEquals(1, successCount.get()); @@ -131,10 +131,10 @@ public class AgentInfoSenderTest { ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory clientFactory = createPinpointClientFactory(); + PinpointClient pinpointClient = createPinpointClient(HOST, PORT, clientFactory); - TcpDataSender dataSender = new TcpDataSender(socket); + TcpDataSender dataSender = new TcpDataSender(pinpointClient); AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo()); try { @@ -145,7 +145,7 @@ public class AgentInfoSenderTest { Thread.sleep(1000L); createAndDeleteServer(serverListener, 5000L); } finally { - closeAll(null, agentInfoSender, socket, socketFactory); + closeAll(null, agentInfoSender, pinpointClient, clientFactory); } assertEquals(1, requestCount.get()); assertEquals(1, successCount.get()); @@ -162,17 +162,17 @@ public class AgentInfoSenderTest { PinpointServerAcceptor serverAcceptor = createServerAcceptor(serverListener); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory socketFactory = createPinpointClientFactory(); + PinpointClient pinpointClient = createPinpointClient(HOST, PORT, socketFactory); - TcpDataSender dataSender = new TcpDataSender(socket); + TcpDataSender dataSender = new TcpDataSender(pinpointClient); AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo()); try { agentInfoSender.start(); Thread.sleep(agentInfoSendRetryIntervalMs * minimumAgentInfoSendRetryCount); } finally { - closeAll(serverAcceptor, agentInfoSender, socket, socketFactory); + closeAll(serverAcceptor, agentInfoSender, pinpointClient, socketFactory); } assertTrue(requestCount.get() >= minimumAgentInfoSendRetryCount); assertEquals(0, successCount.get()); @@ -189,10 +189,10 @@ public class AgentInfoSenderTest { PinpointServerAcceptor serverAcceptor = createServerAcceptor(serverListener); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory clientFactory = createPinpointClientFactory(); + PinpointClient pinpointClient = createPinpointClient(HOST, PORT, clientFactory); - TcpDataSender sender = new TcpDataSender(socket); + TcpDataSender sender = new TcpDataSender(pinpointClient); AgentInfoSender agentInfoSender = new AgentInfoSender(sender, agentInfoSendRetryIntervalMs, getAgentInfo()); final List serverMetaDataObjects = new ArrayList(); serverMetaDataObjects.add(new DefaultServerMetaData("server1", Collections.emptyList(), Collections.emptyMap(), Collections.emptyList())); @@ -207,7 +207,7 @@ public class AgentInfoSenderTest { } Thread.sleep(10000L); } finally { - closeAll(serverAcceptor, agentInfoSender, socket, socketFactory); + closeAll(serverAcceptor, agentInfoSender, pinpointClient, clientFactory); } // Then assertEquals(5, requestCount.get()); @@ -231,10 +231,10 @@ public class AgentInfoSenderTest { PinpointServerAcceptor serverAcceptor = createServerAcceptor(delayedServerListener); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory clientFactory = createPinpointClientFactory(); + PinpointClient pinpointClient = createPinpointClient(HOST, PORT, clientFactory); - TcpDataSender sender = new TcpDataSender(socket); + TcpDataSender sender = new TcpDataSender(pinpointClient); AgentInfoSender agentInfoSender = new AgentInfoSender(sender, agentInfoSendRetryIntervalMs, getAgentInfo()); final ServerMetaDataHolder metaDataContext = new DefaultServerMetaDataHolder(Collections.emptyList()); metaDataContext.addListener(agentInfoSender); @@ -263,7 +263,7 @@ public class AgentInfoSenderTest { try { Thread.sleep(10000L); } finally { - closeAll(serverAcceptor, agentInfoSender, socket, socketFactory); + closeAll(serverAcceptor, agentInfoSender, pinpointClient, clientFactory); } // Then assertTrue("Failed with exceptions : " + exceptions, exceptions.isEmpty()); @@ -281,10 +281,10 @@ public class AgentInfoSenderTest { ResponseServerMessageListener serverListener = new ResponseServerMessageListener(requestCount, successCount, expectedTriesUntilSuccess); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory clientFactory = createPinpointClientFactory(); + PinpointClient pinpointClient = createPinpointClient(HOST, PORT, clientFactory); - TcpDataSender dataSender = new TcpDataSender(socket); + TcpDataSender dataSender = new TcpDataSender(pinpointClient); AgentInfoSender agentInfoSender = new AgentInfoSender(dataSender, agentInfoSendRetryIntervalMs, getAgentInfo()); long startTime = System.currentTimeMillis(); @@ -300,7 +300,7 @@ public class AgentInfoSenderTest { } } finally { - closeAll(null, agentInfoSender, socket, socketFactory); + closeAll(null, agentInfoSender, pinpointClient, clientFactory); } assertEquals(1, successCount.get()); assertEquals(expectedTriesUntilSuccess, requestCount.get()); @@ -328,7 +328,7 @@ public class AgentInfoSenderTest { } } - private void closeAll(PinpointServerAcceptor serverAcceptor, AgentInfoSender agentInfoSender, PinpointSocket socket, PinpointSocketFactory factory) { + private void closeAll(PinpointServerAcceptor serverAcceptor, AgentInfoSender agentInfoSender, PinpointClient pinpointClient, PinpointClientFactory factory) { if (serverAcceptor != null) { serverAcceptor.close(); } @@ -337,8 +337,8 @@ public class AgentInfoSenderTest { agentInfoSender.stop(); } - if (socket != null) { - socket.close(); + if (pinpointClient != null) { + pinpointClient.close(); } if (factory != null) { @@ -410,30 +410,30 @@ public class AgentInfoSenderTest { } } - private PinpointSocketFactory createPinpointSocketFactory() { - PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setTimeoutMillis(1000 * 5); - pinpointSocketFactory.setProperties(Collections.emptyMap()); + private PinpointClientFactory createPinpointClientFactory() { + PinpointClientFactory clientFactory = new PinpointClientFactory(); + clientFactory.setTimeoutMillis(1000 * 5); + clientFactory.setProperties(Collections.emptyMap()); - return pinpointSocketFactory; + return clientFactory; } - private PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) { - PinpointSocket socket = null; + private PinpointClient createPinpointClient(String host, int port, PinpointClientFactory factory) { + PinpointClient pinpointClient = null; for (int i = 0; i < 3; i++) { try { - socket = factory.connect(host, port); + pinpointClient = factory.connect(host, port); logger.info("tcp connect success:{}/{}", host, port); - return socket; + return pinpointClient; } catch (PinpointSocketException e) { logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i); } } logger.warn("change background tcp connect mode {}/{} ", host, port); - socket = factory.scheduledConnect(host, port); + pinpointClient = factory.scheduledConnect(host, port); - return socket; + return pinpointClient; } } diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/context/TraceTest.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/context/TraceTest.java index 624ad3e65..52c769dba 100644 --- a/profiler/src/test/java/com/navercorp/pinpoint/profiler/context/TraceTest.java +++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/context/TraceTest.java @@ -17,17 +17,12 @@ package com.navercorp.pinpoint.profiler.context; import com.navercorp.pinpoint.bootstrap.context.Trace; -import com.navercorp.pinpoint.common.trace.AnnotationKey; -import com.navercorp.pinpoint.common.trace.ServiceType; -import com.navercorp.pinpoint.profiler.context.DefaultTrace; -import com.navercorp.pinpoint.profiler.context.DefaultTraceContext; -import com.navercorp.pinpoint.profiler.context.DefaultTraceId; import com.navercorp.pinpoint.profiler.context.storage.SpanStorage; import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender; import com.navercorp.pinpoint.profiler.sender.LoggingDataSender; import com.navercorp.pinpoint.rpc.FutureListener; import com.navercorp.pinpoint.rpc.ResponseMessage; -import com.navercorp.pinpoint.rpc.client.PinpointSocketReconnectEventListener; +import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener; import com.navercorp.pinpoint.test.TestAgentInformation; import org.apache.thrift.TBase; @@ -104,12 +99,12 @@ public class TraceTest { } @Override - public boolean addReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) { return false; } @Override - public boolean removeReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) { return false; } diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/CountingDataSender.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/CountingDataSender.java index def86fd5d..3c1ecad12 100644 --- a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/CountingDataSender.java +++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/CountingDataSender.java @@ -18,10 +18,9 @@ package com.navercorp.pinpoint.profiler.sender; import com.navercorp.pinpoint.profiler.context.Span; import com.navercorp.pinpoint.profiler.context.SpanChunk; -import com.navercorp.pinpoint.profiler.sender.EnhancedDataSender; import com.navercorp.pinpoint.rpc.FutureListener; import com.navercorp.pinpoint.rpc.ResponseMessage; -import com.navercorp.pinpoint.rpc.client.PinpointSocketReconnectEventListener; +import com.navercorp.pinpoint.rpc.client.PinpointClientReconnectEventListener; import org.apache.thrift.TBase; @@ -59,12 +58,12 @@ public class CountingDataSender implements EnhancedDataSender { } @Override - public boolean addReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener) { return false; } @Override - public boolean removeReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener) { return false; } diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderReconnectTest.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderReconnectTest.java index 6e68b897d..68dbfea45 100644 --- a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderReconnectTest.java +++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderReconnectTest.java @@ -19,13 +19,13 @@ package com.navercorp.pinpoint.profiler.sender; import java.util.Collections; import java.util.Map; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.navercorp.pinpoint.rpc.PinpointSocketException; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseType; import com.navercorp.pinpoint.rpc.packet.PingPacket; @@ -82,10 +82,10 @@ public class TcpDataSenderReconnectTest { public void connectAndSend() throws InterruptedException { PinpointServerAcceptor oldAcceptor = serverAcceptorStart(); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClientFactory clientFactory = createPinpointClientFactory(); + PinpointClient client = createPinpointClient(HOST, PORT, clientFactory); - TcpDataSender sender = new TcpDataSender(socket); + TcpDataSender sender = new TcpDataSender(client); Thread.sleep(500); oldAcceptor.close(); @@ -102,33 +102,33 @@ public class TcpDataSenderReconnectTest { sender.stop(); serverAcceptor.close(); - socket.close(); - socketFactory.release(); + client.close(); + clientFactory.release(); } - private PinpointSocketFactory createPinpointSocketFactory() { - PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setTimeoutMillis(1000 * 5); - pinpointSocketFactory.setProperties(Collections.EMPTY_MAP); + private PinpointClientFactory createPinpointClientFactory() { + PinpointClientFactory clientFactory = new PinpointClientFactory(); + clientFactory.setTimeoutMillis(1000 * 5); + clientFactory.setProperties(Collections.EMPTY_MAP); - return pinpointSocketFactory; + return clientFactory; } - private PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) { - PinpointSocket socket = null; + private PinpointClient createPinpointClient(String host, int port, PinpointClientFactory clientFactory) { + PinpointClient client = null; for (int i = 0; i < 3; i++) { try { - socket = factory.connect(host, port); + client = clientFactory.connect(host, port); logger.info("tcp connect success:{}/{}", host, port); - return socket; + return client; } catch (PinpointSocketException e) { logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i); } } logger.warn("change background tcp connect mode {}/{} ", host, port); - socket = factory.scheduledConnect(host, port); + client = clientFactory.scheduledConnect(host, port); - return socket; + return client; } } diff --git a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderTest.java b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderTest.java index 4b70e08fa..589190b84 100644 --- a/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderTest.java +++ b/profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderTest.java @@ -21,6 +21,8 @@ import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.junit.Assert; import org.junit.After; import org.junit.Before; @@ -29,8 +31,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.navercorp.pinpoint.rpc.PinpointSocketException; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseType; import com.navercorp.pinpoint.rpc.packet.PingPacket; @@ -96,11 +96,11 @@ public class TcpDataSenderTest { public void connectAndSend() throws InterruptedException { this.sendLatch = new CountDownLatch(2); - PinpointSocketFactory socketFactory = createPinpointSocketFactory(); + PinpointClientFactory clientFactory = createPinpointClientFactory(); - PinpointSocket socket = createPinpointSocket(HOST, PORT, socketFactory); + PinpointClient client = createPinpointClient(HOST, PORT, clientFactory); - TcpDataSender sender = new TcpDataSender(socket); + TcpDataSender sender = new TcpDataSender(client); try { sender.send(new TApiMetaData("test", System.currentTimeMillis(), 1, "TestApi")); sender.send(new TApiMetaData("test", System.currentTimeMillis(), 1, "TestApi")); @@ -111,38 +111,38 @@ public class TcpDataSenderTest { } finally { sender.stop(); - if (socket != null) { - socket.close(); + if (client != null) { + client.close(); } - if (socketFactory != null) { - socketFactory.release(); + if (clientFactory != null) { + clientFactory.release(); } } } - private PinpointSocketFactory createPinpointSocketFactory() { - PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setTimeoutMillis(1000 * 5); - pinpointSocketFactory.setProperties(Collections.EMPTY_MAP); + private PinpointClientFactory createPinpointClientFactory() { + PinpointClientFactory clientFactory = new PinpointClientFactory(); + clientFactory.setTimeoutMillis(1000 * 5); + clientFactory.setProperties(Collections.EMPTY_MAP); - return pinpointSocketFactory; + return clientFactory; } - private PinpointSocket createPinpointSocket(String host, int port, PinpointSocketFactory factory) { - PinpointSocket socket = null; + private PinpointClient createPinpointClient(String host, int port, PinpointClientFactory clientFactory) { + PinpointClient client = null; for (int i = 0; i < 3; i++) { try { - socket = factory.connect(host, port); + client = clientFactory.connect(host, port); logger.info("tcp connect success:{}/{}", host, port); - return socket; + return client; } catch (PinpointSocketException e) { logger.warn("tcp connect fail:{}/{} try reconnect, retryCount:{}", host, port, i); } } logger.warn("change background tcp connect mode {}/{} ", host, port); - socket = factory.scheduledConnect(host, port); + client = clientFactory.scheduledConnect(host, port); - return socket; + return client; } } diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/ClassPreLoader.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/ClassPreLoader.java index a3c47240a..6dfd72bb9 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/ClassPreLoader.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/ClassPreLoader.java @@ -19,8 +19,8 @@ package com.navercorp.pinpoint.rpc; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import com.navercorp.pinpoint.rpc.server.PinpointServerAcceptor; /** @@ -39,15 +39,15 @@ public final class ClassPreLoader { public static void preload(int port) { PinpointServerAcceptor serverAcceptor = null; - PinpointSocket socket = null; - PinpointSocketFactory socketFactory = null; + PinpointClient client = null; + PinpointClientFactory clientFactory = null; try { serverAcceptor = new PinpointServerAcceptor(); serverAcceptor.bind("127.0.0.1", port); - socketFactory = new PinpointSocketFactory(); - socket = socketFactory.connect("127.0.0.1", port); - socket.sendSync(new byte[0]); + clientFactory = new PinpointClientFactory(); + client = clientFactory.connect("127.0.0.1", port); + client.sendSync(new byte[0]); } catch (Exception ex) { @@ -63,17 +63,17 @@ public final class ClassPreLoader { throw new PinpointSocketException(ex.getMessage(), ex); } } finally { - if (socket != null) { + if (client != null) { try { - socket.close(); + client.close(); } catch (Exception e) { e.printStackTrace(); } } - if(socketFactory != null) { + if(clientFactory != null) { try { - socketFactory.release(); + clientFactory.release(); } catch (Exception e) { e.printStackTrace(); } diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DefaultPinpointClientHandler.java similarity index 91% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DefaultPinpointClientHandler.java index 2909b8ca2..5cc561a4d 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DefaultPinpointClientHandler.java @@ -71,7 +71,7 @@ import com.navercorp.pinpoint.rpc.util.TimerFactory; * @author netspider * @author koo.taejin */ -public class PinpointSocketHandler extends SimpleChannelHandler implements SocketHandler { +public class DefaultPinpointClientHandler extends SimpleChannelHandler implements PinpointClientHandler { private static final long DEFAULT_PING_DELAY = 60 * 1000 * 5; private static final long DEFAULT_TIMEOUTMILLIS = 3 * 1000; @@ -83,7 +83,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke private final int socketId; private final AtomicInteger pingIdGenerator; - private final PinpointSocketHandlerState state; + private final PinpointClientHandlerState state; private volatile Channel channel; @@ -95,9 +95,9 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke private final Timer channelTimer; - private final PinpointSocketFactory pinpointSocketFactory; + private final PinpointClientFactory pinpointClientFactory; private SocketAddress connectSocketAddress; - private volatile PinpointSocket pinpointSocket; + private volatile PinpointClient pinpointClient; private final MessageListener messageListener; private final ServerStreamChannelMessageListener serverStreamChannelMessageListener; @@ -109,46 +109,46 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke private final ChannelFutureListener sendClosePacketFailFutureListener = new WriteFailFutureListener(this.logger, "sendClosedPacket() write fail.", "sendClosedPacket() write success."); - private final PinpointClientSocketHandshaker handshaker; + private final PinpointClientHandshaker handshaker; private final ConnectFuture connectFuture = new ConnectFuture(); private final String objectUniqName; - public PinpointSocketHandler(PinpointSocketFactory pinpointSocketFactory) { - this(pinpointSocketFactory, DEFAULT_PING_DELAY, DEFAULT_ENABLE_WORKER_PACKET_DELAY, DEFAULT_TIMEOUTMILLIS); + public DefaultPinpointClientHandler(PinpointClientFactory pinpointClientFactory) { + this(pinpointClientFactory, DEFAULT_PING_DELAY, DEFAULT_ENABLE_WORKER_PACKET_DELAY, DEFAULT_TIMEOUTMILLIS); } - public PinpointSocketHandler(PinpointSocketFactory pinpointSocketFactory, long pingDelay, long handshakeRetryInterval, long timeoutMillis) { - if (pinpointSocketFactory == null) { - throw new NullPointerException("pinpointSocketFactory must not be null"); + public DefaultPinpointClientHandler(PinpointClientFactory pinpointClientFactory, long pingDelay, long handshakeRetryInterval, long timeoutMillis) { + if (pinpointClientFactory == null) { + throw new NullPointerException("pinpointClientFactory must not be null"); } - HashedWheelTimer timer = TimerFactory.createHashedWheelTimer("Pinpoint-SocketHandler-Timer", 100, TimeUnit.MILLISECONDS, 512); + HashedWheelTimer timer = TimerFactory.createHashedWheelTimer("Pinpoint-PinpointClientHandler-Timer", 100, TimeUnit.MILLISECONDS, 512); timer.start(); this.channelTimer = timer; - this.pinpointSocketFactory = pinpointSocketFactory; + this.pinpointClientFactory = pinpointClientFactory; this.requestManager = new RequestManager(timer, timeoutMillis); this.pingDelay = pingDelay; this.timeoutMillis = timeoutMillis; - this.messageListener = pinpointSocketFactory.getMessageListener(SimpleLoggingMessageListener.LISTENER); - this.serverStreamChannelMessageListener = pinpointSocketFactory.getServerStreamChannelMessageListener(DisabledServerStreamChannelMessageListener.INSTANCE); + this.messageListener = pinpointClientFactory.getMessageListener(SimpleLoggingMessageListener.LISTENER); + this.serverStreamChannelMessageListener = pinpointClientFactory.getServerStreamChannelMessageListener(DisabledServerStreamChannelMessageListener.INSTANCE); this.objectUniqName = ClassUtils.simpleClassNameAndHashCodeString(this); - this.handshaker = new PinpointClientSocketHandshaker(channelTimer, (int) handshakeRetryInterval, maxHandshakeCount); + this.handshaker = new PinpointClientHandshaker(channelTimer, (int) handshakeRetryInterval, maxHandshakeCount); - this.socketId = pinpointSocketFactory.issueNewSocketId(); + this.socketId = pinpointClientFactory.issueNewSocketId(); this.pingIdGenerator = new AtomicInteger(0); - this.state = new PinpointSocketHandlerState(this.objectUniqName); + this.state = new PinpointClientHandlerState(this.objectUniqName); } - public void setPinpointSocket(PinpointSocket pinpointSocket) { - if (pinpointSocket == null) { - throw new NullPointerException("pinpointSocket must not be null"); + public void setPinpointClient(PinpointClient pinpointClient) { + if (pinpointClient == null) { + throw new NullPointerException("pinpointClient must not be null"); } - this.pinpointSocket = pinpointSocket; + this.pinpointClient = pinpointClient; } public void setConnectSocketAddress(SocketAddress connectSocketAddress) { @@ -191,7 +191,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke registerPing(); Map handshakeData = new HashMap(); - handshakeData.putAll(pinpointSocketFactory.getProperties()); + handshakeData.putAll(pinpointClientFactory.getProperties()); handshakeData.put("socketId", socketId); handshaker.handshakeStart(channel, handshakeData); @@ -204,7 +204,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke private void prepareChannel(Channel channel) { StreamChannelManager streamChannelManager = new StreamChannelManager(channel, IDGenerator.createOddIdGenerator(), serverStreamChannelMessageListener); - PinpointSocketHandlerContext context = new PinpointSocketHandlerContext(channel, streamChannelManager); + PinpointClientHandlerContext context = new PinpointClientHandlerContext(channel, streamChannelManager); channel.setAttachment(context); } @@ -356,7 +356,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke public ClientStreamChannelContext createStreamChannel(byte[] payload, ClientStreamChannelMessageListener clientStreamChannelMessageListener) { ensureOpen(); - PinpointSocketHandlerContext context = getChannelContext(channel); + PinpointClientHandlerContext context = getChannelContext(channel); return context.createStream(payload, clientStreamChannelMessageListener); } @@ -364,7 +364,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke public StreamChannelContext findStreamChannel(int streamChannelId) { ensureOpen(); - PinpointSocketHandlerContext context = getChannelContext(channel); + PinpointClientHandlerContext context = getChannelContext(channel); return context.getStreamChannel(streamChannelId); } @@ -392,7 +392,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke case PacketType.APPLICATION_STREAM_RESPONSE: case PacketType.APPLICATION_STREAM_PING: case PacketType.APPLICATION_STREAM_PONG: - PinpointSocketHandlerContext context = getChannelContext(channel); + PinpointClientHandlerContext context = getChannelContext(channel); context.handleStreamEvent((StreamPacket) message); return; case PacketType.CONTROL_SERVER_CLOSE: @@ -477,7 +477,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke throw new PinpointSocketException("Invalid socket state:" + currentStateCode); } - // Calling this method on a closed SocketHandler has no effect. + // Calling this method on a closed PinpointClientHandler has no effect. public void close() { logger.debug("{} close() started.", objectUniqName); @@ -507,7 +507,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke } } - // Calling this method on a closed SocketHandler has no effect. + // Calling this method on a closed PinpointClientHandler has no effect. private void closeResources() { logger.debug("{} closeResources() started.", objectUniqName); @@ -523,7 +523,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke } // stream channel clear and send stream close packet - PinpointSocketHandlerContext context = getChannelContext(channel); + PinpointClientHandlerContext context = getChannelContext(channel); if (context != null) { context.closeAllStreamChannel(); } @@ -547,7 +547,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke logger.info("{} channelClosed() started.", objectUniqName); try { - boolean factoryReleased = pinpointSocketFactory.isReleased(); + boolean factoryReleased = pinpointClientFactory.isReleased(); boolean needReconnect = false; SocketStateCode currentStateCode = state.getCurrentStateCode(); @@ -566,7 +566,7 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke } if (needReconnect) { - pinpointSocketFactory.reconnect(this.pinpointSocket, this.connectSocketAddress); + pinpointClientFactory.reconnect(this.pinpointClient, this.connectSocketAddress); } } finally { closeResources(); @@ -601,11 +601,11 @@ public class PinpointSocketHandler extends SimpleChannelHandler implements Socke return state.getCurrentStateCode(); } - private PinpointSocketHandlerContext getChannelContext(Channel channel) { + private PinpointClientHandlerContext getChannelContext(Channel channel) { if (channel == null) { throw new NullPointerException("channel must not be null"); } - return (PinpointSocketHandlerContext) channel.getAttachment(); + return (PinpointClientHandlerContext) channel.getAttachment(); } @Override diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointSocketReconnectEventListener.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointClientReconnectEventListener.java similarity index 80% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointSocketReconnectEventListener.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointClientReconnectEventListener.java index 948ce1271..3e89d74fe 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointSocketReconnectEventListener.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointClientReconnectEventListener.java @@ -16,10 +16,10 @@ package com.navercorp.pinpoint.rpc.client; -public class DummyPinpointSocketReconnectEventListener implements PinpointSocketReconnectEventListener { +public class DummyPinpointClientReconnectEventListener implements PinpointClientReconnectEventListener { @Override - public void reconnectPerformed(PinpointSocket socket) { + public void reconnectPerformed(PinpointClient client) { } diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocket.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClient.java similarity index 61% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocket.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClient.java index 86a4e4063..032108909 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocket.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClient.java @@ -37,38 +37,38 @@ import com.navercorp.pinpoint.rpc.util.AssertUtils; * @author koo.taejin * @author netspider */ -public class PinpointSocket { +public class PinpointClient { private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private volatile SocketHandler socketHandler; + private volatile PinpointClientHandler pinpointClientHandler; private volatile boolean closed; - private List reconnectEventListeners = new CopyOnWriteArrayList(); + private List reconnectEventListeners = new CopyOnWriteArrayList(); - public PinpointSocket() { - this(new ReconnectStateSocketHandler()); + public PinpointClient() { + this(new ReconnectStateClientHandler()); } - public PinpointSocket(SocketHandler socketHandler) { - AssertUtils.assertNotNull(socketHandler, "socketHandler"); + public PinpointClient(PinpointClientHandler pinpointClientHandler) { + AssertUtils.assertNotNull(pinpointClientHandler, "pinpointClientHandler"); - this.socketHandler = socketHandler; - socketHandler.setPinpointSocket(this); + this.pinpointClientHandler = pinpointClientHandler; + pinpointClientHandler.setPinpointClient(this); } - void reconnectSocketHandler(SocketHandler socketHandler) { - AssertUtils.assertNotNull(socketHandler, "socketHandler"); + void reconnectSocketHandler(PinpointClientHandler pinpointClientHandler) { + AssertUtils.assertNotNull(pinpointClientHandler, "pinpointClientHandler"); if (closed) { - logger.warn("reconnectSocketHandler(). socketHandler force close."); - socketHandler.close(); + logger.warn("reconnectClientHandler(). pinpointClientHandler force close."); + pinpointClientHandler.close(); return; } - logger.warn("reconnectSocketHandler:{}", socketHandler); + logger.warn("reconnectClientHandler:{}", pinpointClientHandler); - this.socketHandler = socketHandler; + this.pinpointClientHandler = pinpointClientHandler; notifyReconnectEvent(); } @@ -78,7 +78,7 @@ public class PinpointSocket { because reconnectEventListener's constructor contains Dummy and can't be access through setter, guarantee it is not null. */ - public boolean addPinpointSocketReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean addPinpointClientReconnectEventListener(PinpointClientReconnectEventListener eventListener) { if (eventListener == null) { return false; } @@ -86,7 +86,7 @@ public class PinpointSocket { return this.reconnectEventListeners.add(eventListener); } - public boolean removePinpointSocketReconnectEventListener(PinpointSocketReconnectEventListener eventListener) { + public boolean removePinpointClientReconnectEventListener(PinpointClientReconnectEventListener eventListener) { if (eventListener == null) { return false; } @@ -95,56 +95,56 @@ public class PinpointSocket { } private void notifyReconnectEvent() { - for (PinpointSocketReconnectEventListener eachListener : this.reconnectEventListeners) { + for (PinpointClientReconnectEventListener eachListener : this.reconnectEventListeners) { eachListener.reconnectPerformed(this); } } public void sendSync(byte[] bytes) { ensureOpen(); - socketHandler.sendSync(bytes); + pinpointClientHandler.sendSync(bytes); } public Future sendAsync(byte[] bytes) { ensureOpen(); - return socketHandler.sendAsync(bytes); + return pinpointClientHandler.sendAsync(bytes); } public void send(byte[] bytes) { ensureOpen(); - socketHandler.send(bytes); + pinpointClientHandler.send(bytes); } public Future request(byte[] bytes) { - if (socketHandler == null) { + if (pinpointClientHandler == null) { return returnFailureFuture(); } - return socketHandler.request(bytes); + return pinpointClientHandler.request(bytes); } public ClientStreamChannelContext createStreamChannel(byte[] payload, ClientStreamChannelMessageListener clientStreamChannelMessageListener) { // StreamChannel must be changed into interface in order to throw the StreamChannel that returns failure. // fow now throw just exception ensureOpen(); - return socketHandler.createStreamChannel(payload, clientStreamChannelMessageListener); + return pinpointClientHandler.createStreamChannel(payload, clientStreamChannelMessageListener); } public StreamChannelContext findStreamChannel(int streamChannelId) { ensureOpen(); - return socketHandler.findStreamChannel(streamChannelId); + return pinpointClientHandler.findStreamChannel(streamChannelId); } private Future returnFailureFuture() { DefaultFuture future = new DefaultFuture(); - future.setFailure(new PinpointSocketException("socketHandler is null")); + future.setFailure(new PinpointSocketException("pinpointClientHandler is null")); return future; } private void ensureOpen() { - if (socketHandler == null) { - throw new PinpointSocketException("socketHandler is null"); + if (pinpointClientHandler == null) { + throw new PinpointSocketException("pinpointClientHandler is null"); } } @@ -154,11 +154,11 @@ public class PinpointSocket { * */ public void sendPing() { - SocketHandler socketHandler = this.socketHandler; - if (socketHandler == null) { + PinpointClientHandler pinpointClientHandler = this.pinpointClientHandler; + if (pinpointClientHandler == null) { return; } - socketHandler.sendPing(); + pinpointClientHandler.sendPing(); } public void close() { @@ -168,11 +168,11 @@ public class PinpointSocket { } closed = true; } - SocketHandler socketHandler = this.socketHandler; - if (socketHandler == null) { + PinpointClientHandler pinpointClientHandler = this.pinpointClientHandler; + if (pinpointClientHandler == null) { return; } - socketHandler.close(); + pinpointClientHandler.close(); } public boolean isClosed() { @@ -180,6 +180,6 @@ public class PinpointSocket { } public boolean isConnected() { - return this.socketHandler.isConnected(); + return this.pinpointClientHandler.isConnected(); } } diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactory.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactory.java similarity index 82% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactory.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactory.java index 3899f368d..f0e283579 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactory.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactory.java @@ -56,7 +56,7 @@ import com.navercorp.pinpoint.rpc.util.TimerFactory; * @author emeroad * @author koo.taejin */ -public class PinpointSocketFactory { +public class PinpointClientFactory { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @@ -89,11 +89,11 @@ public class PinpointSocketFactory { LoggerFactorySetup.setupSlf4jLoggerFactory(); } - public PinpointSocketFactory() { + public PinpointClientFactory() { this(1, 1); } - public PinpointSocketFactory(int bossCount, int workerCount) { + public PinpointClientFactory(int bossCount, int workerCount) { if (bossCount < 1) { throw new IllegalArgumentException("bossCount is negative: " + bossCount); } @@ -115,8 +115,8 @@ public class PinpointSocketFactory { } private void addPipeline(ClientBootstrap bootstrap) { - SocketClientPipelineFactory socketClientPipelineFactory = new SocketClientPipelineFactory(this); - bootstrap.setPipelineFactory(socketClientPipelineFactory); + PinpointClientPipelineFactory pinpointClientPipelineFactory = new PinpointClientPipelineFactory(this); + bootstrap.setPipelineFactory(pinpointClientPipelineFactory); } private void setOptions(ClientBootstrap bootstrap) { @@ -204,57 +204,57 @@ public class PinpointSocketFactory { return new NioClientSocketChannelFactory(bossPool, workerPool); } - public PinpointSocket connect(String host, int port) throws PinpointSocketException { + public PinpointClient connect(String host, int port) throws PinpointSocketException { SocketAddress address = new InetSocketAddress(host, port); ChannelFuture connectFuture = bootstrap.connect(address); - SocketHandler socketHandler = getSocketHandler(connectFuture, address); + PinpointClientHandler pinpointClientHandler = getSocketHandler(connectFuture, address); - PinpointSocket pinpointSocket = new PinpointSocket(socketHandler); - traceSocket(pinpointSocket); - return pinpointSocket; + PinpointClient pinpointClient = new PinpointClient(pinpointClientHandler); + traceSocket(pinpointClient); + return pinpointClient; } - public PinpointSocket reconnect(String host, int port) throws PinpointSocketException { + public PinpointClient reconnect(String host, int port) throws PinpointSocketException { SocketAddress address = new InetSocketAddress(host, port); ChannelFuture connectFuture = bootstrap.connect(address); - SocketHandler socketHandler = getSocketHandler(connectFuture, address); + PinpointClientHandler pinpointClientHandler = getSocketHandler(connectFuture, address); - PinpointSocket pinpointSocket = new PinpointSocket(socketHandler); - traceSocket(pinpointSocket); - return pinpointSocket; + PinpointClient pinpointClient = new PinpointClient(pinpointClientHandler); + traceSocket(pinpointClient); + return pinpointClient; } /* trace mechanism is needed in case of calling close without closing socket it is okay to make that later because this is a exceptional case. */ - private void traceSocket(PinpointSocket pinpointSocket) { + private void traceSocket(PinpointClient pinpointClient) { } - public PinpointSocket scheduledConnect(String host, int port) { - PinpointSocket pinpointSocket = new PinpointSocket(new ReconnectStateSocketHandler()); + public PinpointClient scheduledConnect(String host, int port) { + PinpointClient pinpointClient = new PinpointClient(new ReconnectStateClientHandler()); SocketAddress address = new InetSocketAddress(host, port); - reconnect(pinpointSocket, address); - return pinpointSocket; + reconnect(pinpointClient, address); + return pinpointClient; } - SocketHandler getSocketHandler(ChannelFuture channelConnectFuture, SocketAddress address) { + PinpointClientHandler getSocketHandler(ChannelFuture channelConnectFuture, SocketAddress address) { if (address == null) { throw new NullPointerException("address"); } - SocketHandler socketHandler = getSocketHandler(channelConnectFuture.getChannel()); - socketHandler.setConnectSocketAddress(address); + PinpointClientHandler pinpointClientHandler = getSocketHandler(channelConnectFuture.getChannel()); + pinpointClientHandler.setConnectSocketAddress(address); - ConnectFuture handlerConnectFuture = socketHandler.getConnectFuture(); + ConnectFuture handlerConnectFuture = pinpointClientHandler.getConnectFuture(); handlerConnectFuture.awaitUninterruptibly(); if (ConnectFuture.Result.FAIL == handlerConnectFuture.getResult()) { throw new PinpointSocketException("connect fail to " + address + ".", channelConnectFuture.getCause()); } - return socketHandler; + return pinpointClientHandler; } public ChannelFuture reconnect(final SocketAddress remoteAddress) { @@ -269,8 +269,8 @@ public class PinpointSocketFactory { } catch (Exception e) { throw new ChannelPipelineException("Failed to initialize a pipeline.", e); } - SocketHandler socketHandler = (PinpointSocketHandler) pipeline.getLast(); - socketHandler.initReconnect(); + PinpointClientHandler pinpointClientHandler = (DefaultPinpointClientHandler) pipeline.getLast(); + pinpointClientHandler.initReconnect(); // Set the options. @@ -294,30 +294,30 @@ public class PinpointSocketFactory { } - private SocketHandler getSocketHandler(Channel channel) { - return (SocketHandler) channel.getPipeline().getLast(); + private PinpointClientHandler getSocketHandler(Channel channel) { + return (PinpointClientHandler) channel.getPipeline().getLast(); } - void reconnect(final PinpointSocket pinpointSocket, final SocketAddress socketAddress) { - ConnectEvent connectEvent = new ConnectEvent(pinpointSocket, socketAddress); + void reconnect(final PinpointClient pinpointClient, final SocketAddress socketAddress) { + ConnectEvent connectEvent = new ConnectEvent(pinpointClient, socketAddress); timer.newTimeout(connectEvent, reconnectDelay, TimeUnit.MILLISECONDS); } private class ConnectEvent implements TimerTask { private final Logger logger = LoggerFactory.getLogger(getClass()); - private final PinpointSocket pinpointSocket; + private final PinpointClient pinpointClient; private final SocketAddress socketAddress; - private ConnectEvent(PinpointSocket pinpointSocket, SocketAddress socketAddress) { - if (pinpointSocket == null) { - throw new NullPointerException("pinpointSocket must not be null"); + private ConnectEvent(PinpointClient pinpointClient, SocketAddress socketAddress) { + if (pinpointClient == null) { + throw new NullPointerException("pinpointClient must not be null"); } if (socketAddress == null) { throw new NullPointerException("socketAddress must not be null"); } - this.pinpointSocket = pinpointSocket; + this.pinpointClient = pinpointClient; this.socketAddress = socketAddress; } @@ -327,18 +327,18 @@ public class PinpointSocketFactory { return; } - // Just return not to try reconnection when event has been fired but pinpointSocket already closed. - if (pinpointSocket.isClosed()) { - logger.debug("pinpointSocket is already closed."); + // Just return not to try reconnection when event has been fired but pinpointClient already closed. + if (pinpointClient.isClosed()) { + logger.debug("pinpointClient is already closed."); return; } logger.warn("try reconnect. connectAddress:{}", socketAddress); final ChannelFuture channelFuture = reconnect(socketAddress); Channel channel = channelFuture.getChannel(); - final SocketHandler socketHandler = getSocketHandler(channel); - socketHandler.setConnectSocketAddress(socketAddress); - socketHandler.setPinpointSocket(pinpointSocket); + final PinpointClientHandler pinpointClientHandler = getSocketHandler(channel); + pinpointClientHandler.setConnectSocketAddress(socketAddress); + pinpointClientHandler.setPinpointClient(pinpointClient); channelFuture.addListener(new ChannelFutureListener() { @Override @@ -346,9 +346,9 @@ public class PinpointSocketFactory { if (future.isSuccess()) { Channel channel = future.getChannel(); logger.warn("reconnect success {}, {}", socketAddress, channel); - pinpointSocket.reconnectSocketHandler(socketHandler); + pinpointClient.reconnectSocketHandler(pinpointClientHandler); } else { - if (!pinpointSocket.isClosed()) { + if (!pinpointClient.isClosed()) { /* // comment out because exception message can be taken at exceptionCaught @@ -357,9 +357,9 @@ public class PinpointSocketFactory { logger.warn("reconnect fail. {} Caused:{}", socketAddress, cause.getMessage()); } */ - reconnect(pinpointSocket, socketAddress); + reconnect(pinpointClient, socketAddress); } else { - logger.info("pinpointSocket is closed. stop reconnect."); + logger.info("pinpointClient is closed. stop reconnect."); } } } diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketHandler.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandler.java similarity index 94% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketHandler.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandler.java index f29f859d3..b7802ea80 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketHandler.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandler.java @@ -29,7 +29,7 @@ import com.navercorp.pinpoint.rpc.stream.StreamChannelContext; * @author emeroad * @author netspider */ -public interface SocketHandler { +public interface PinpointClientHandler { void setConnectSocketAddress(SocketAddress address); @@ -37,7 +37,7 @@ public interface SocketHandler { ConnectFuture getConnectFuture(); - void setPinpointSocket(PinpointSocket pinpointSocket); + void setPinpointClient(PinpointClient pinpointClient); void sendSync(byte[] bytes); diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerContext.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerContext.java similarity index 92% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerContext.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerContext.java index 83eec5477..1b4ce81d7 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerContext.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerContext.java @@ -1,65 +1,65 @@ -/* - * Copyright 2014 NAVER Corp. - * - * Licensed 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 com.navercorp.pinpoint.rpc.client; - -import org.jboss.netty.channel.Channel; - -import com.navercorp.pinpoint.rpc.packet.stream.StreamPacket; -import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelContext; -import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelMessageListener; -import com.navercorp.pinpoint.rpc.stream.StreamChannelContext; -import com.navercorp.pinpoint.rpc.stream.StreamChannelManager; - -/** - * @author Taejin Koo - */ -public class PinpointSocketHandlerContext { - private final Channel channel; - private final StreamChannelManager streamChannelManager; - - public PinpointSocketHandlerContext(Channel channel, StreamChannelManager streamChannelManager) { - if (channel == null) { - throw new NullPointerException("channel must not be null"); - } - if (streamChannelManager == null) { - throw new NullPointerException("streamChannelManager must not be null"); - } - this.channel = channel; - this.streamChannelManager = streamChannelManager; - } - - public Channel getChannel() { - return channel; - } - - public ClientStreamChannelContext createStream(byte[] payload, ClientStreamChannelMessageListener clientStreamChannelMessageListener) { - return streamChannelManager.openStreamChannel(payload, clientStreamChannelMessageListener); - } - - public void handleStreamEvent(StreamPacket message) { - streamChannelManager.messageReceived(message); - } - - public void closeAllStreamChannel() { - streamChannelManager.close(); - } - - public StreamChannelContext getStreamChannel(int streamChannelId) { - return streamChannelManager.findStreamChannel(streamChannelId); - } - -} +/* + * Copyright 2014 NAVER Corp. + * + * Licensed 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 com.navercorp.pinpoint.rpc.client; + +import org.jboss.netty.channel.Channel; + +import com.navercorp.pinpoint.rpc.packet.stream.StreamPacket; +import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelContext; +import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelMessageListener; +import com.navercorp.pinpoint.rpc.stream.StreamChannelContext; +import com.navercorp.pinpoint.rpc.stream.StreamChannelManager; + +/** + * @author Taejin Koo + */ +public class PinpointClientHandlerContext { + private final Channel channel; + private final StreamChannelManager streamChannelManager; + + public PinpointClientHandlerContext(Channel channel, StreamChannelManager streamChannelManager) { + if (channel == null) { + throw new NullPointerException("channel must not be null"); + } + if (streamChannelManager == null) { + throw new NullPointerException("streamChannelManager must not be null"); + } + this.channel = channel; + this.streamChannelManager = streamChannelManager; + } + + public Channel getChannel() { + return channel; + } + + public ClientStreamChannelContext createStream(byte[] payload, ClientStreamChannelMessageListener clientStreamChannelMessageListener) { + return streamChannelManager.openStreamChannel(payload, clientStreamChannelMessageListener); + } + + public void handleStreamEvent(StreamPacket message) { + streamChannelManager.messageReceived(message); + } + + public void closeAllStreamChannel() { + streamChannelManager.close(); + } + + public StreamChannelContext getStreamChannel(int streamChannelId) { + return streamChannelManager.findStreamChannel(streamChannelId); + } + +} diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerState.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerState.java similarity index 94% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerState.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerState.java index acafe7ead..bc322d0ea 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerState.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerState.java @@ -1,169 +1,169 @@ -/* - * Copyright 2014 NAVER Corp. - * - * Licensed 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 com.navercorp.pinpoint.rpc.client; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.navercorp.pinpoint.rpc.common.SocketState; -import com.navercorp.pinpoint.rpc.common.SocketStateChangeResult; -import com.navercorp.pinpoint.rpc.common.SocketStateCode; - -/** - * @author Taejin Koo - */ -public class PinpointSocketHandlerState { - - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - - private final String objectUniqName; - private final SocketState state; - - public PinpointSocketHandlerState(String objectUniqName) { - this.objectUniqName = objectUniqName; - this.state = new SocketState(); - } - - SocketStateChangeResult toBeingConnect() { - SocketStateCode nextState = SocketStateCode.BEING_CONNECT; - return to(nextState); - } - - SocketStateChangeResult toConnected() { - SocketStateCode nextState = SocketStateCode.CONNECTED; - return to(nextState); - } - - SocketStateChangeResult toConnectFailed() { - SocketStateCode nextState = SocketStateCode.CONNECT_FAILED; - return to(nextState); - } - - SocketStateChangeResult toRunWithoutHandshake() { - SocketStateCode nextState = SocketStateCode.RUN_WITHOUT_HANDSHAKE; - return to(nextState); - } - - SocketStateChangeResult toRunSimplex() { - SocketStateCode nextState = SocketStateCode.RUN_SIMPLEX; - return to(nextState); - } - - SocketStateChangeResult toRunDuplex() { - SocketStateCode nextState = SocketStateCode.RUN_DUPLEX; - return to(nextState); - } - - SocketStateChangeResult toBeingClose() { - SocketStateCode nextState = SocketStateCode.BEING_CLOSE_BY_CLIENT; - return to(nextState); - } - - SocketStateChangeResult toBeingCloseByPeer() { - SocketStateCode nextState = SocketStateCode.BEING_CLOSE_BY_SERVER; - return to(nextState); - } - - SocketStateChangeResult toClosed() { - SocketStateCode nextState = SocketStateCode.CLOSED_BY_CLIENT; - return to(nextState); - } - - SocketStateChangeResult toClosedByPeer() { - SocketStateCode nextState = SocketStateCode.CLOSED_BY_SERVER; - return to(nextState); - } - - SocketStateChangeResult toUnexpectedClosed() { - SocketStateCode nextState = SocketStateCode.UNEXPECTED_CLOSE_BY_CLIENT; - return to(nextState); - } - - SocketStateChangeResult toUnexpectedClosedByPeer() { - SocketStateCode nextState = SocketStateCode.UNEXPECTED_CLOSE_BY_SERVER; - return to(nextState); - } - - SocketStateChangeResult toErrorUnknown() { - SocketStateCode nextState = SocketStateCode.ERROR_UNKNOWN; - return to(nextState); - } - - private SocketStateChangeResult to(SocketStateCode nextState) { - logger.debug("{} stateTo() started. to:{}", objectUniqName, nextState); - - SocketStateChangeResult stateChangeResult = state.changeState(nextState); - - logger.info("{} stateTo() completed. {}", objectUniqName, stateChangeResult); - - return stateChangeResult; - } - - boolean isBeforeConnected(SocketStateCode currentStateCode) { - return SocketStateCode.isBeforeConnected(currentStateCode); - } - - boolean isEnableCommunication() { - return SocketStateCode.isRun(getCurrentStateCode()); - } - - boolean isEnableCommunication(SocketStateCode currentStateCode) { - return SocketStateCode.isRun(currentStateCode); - } - - boolean isEnableDuplexCommunication() { - return SocketStateCode.isRunDuplex(getCurrentStateCode()); - } - - boolean isClosed() { - return SocketStateCode.isClosed(getCurrentStateCode()); - } - - boolean isClosed(SocketStateCode currentStateCode) { - return SocketStateCode.isClosed(currentStateCode); - } - - boolean onClose(SocketStateCode currentStateCode) { - return SocketStateCode.onClose(currentStateCode); - } - - boolean isReconnect(SocketStateCode currentStateCode) { - if (currentStateCode == SocketStateCode.BEING_CLOSE_BY_SERVER) { - return true; - } - - if (currentStateCode == SocketStateCode.CLOSED_BY_SERVER) { - return true; - } - - if (currentStateCode == SocketStateCode.UNEXPECTED_CLOSE_BY_SERVER) { - return true; - } - - return false; - } - - SocketStateCode getCurrentStateCode() { - return state.getCurrentState(); - } - - @Override - public String toString() { - return state.toString(); - } - -} +/* + * Copyright 2014 NAVER Corp. + * + * Licensed 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 com.navercorp.pinpoint.rpc.client; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.navercorp.pinpoint.rpc.common.SocketState; +import com.navercorp.pinpoint.rpc.common.SocketStateChangeResult; +import com.navercorp.pinpoint.rpc.common.SocketStateCode; + +/** + * @author Taejin Koo + */ +public class PinpointClientHandlerState { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + private final String objectUniqName; + private final SocketState state; + + public PinpointClientHandlerState(String objectUniqName) { + this.objectUniqName = objectUniqName; + this.state = new SocketState(); + } + + SocketStateChangeResult toBeingConnect() { + SocketStateCode nextState = SocketStateCode.BEING_CONNECT; + return to(nextState); + } + + SocketStateChangeResult toConnected() { + SocketStateCode nextState = SocketStateCode.CONNECTED; + return to(nextState); + } + + SocketStateChangeResult toConnectFailed() { + SocketStateCode nextState = SocketStateCode.CONNECT_FAILED; + return to(nextState); + } + + SocketStateChangeResult toRunWithoutHandshake() { + SocketStateCode nextState = SocketStateCode.RUN_WITHOUT_HANDSHAKE; + return to(nextState); + } + + SocketStateChangeResult toRunSimplex() { + SocketStateCode nextState = SocketStateCode.RUN_SIMPLEX; + return to(nextState); + } + + SocketStateChangeResult toRunDuplex() { + SocketStateCode nextState = SocketStateCode.RUN_DUPLEX; + return to(nextState); + } + + SocketStateChangeResult toBeingClose() { + SocketStateCode nextState = SocketStateCode.BEING_CLOSE_BY_CLIENT; + return to(nextState); + } + + SocketStateChangeResult toBeingCloseByPeer() { + SocketStateCode nextState = SocketStateCode.BEING_CLOSE_BY_SERVER; + return to(nextState); + } + + SocketStateChangeResult toClosed() { + SocketStateCode nextState = SocketStateCode.CLOSED_BY_CLIENT; + return to(nextState); + } + + SocketStateChangeResult toClosedByPeer() { + SocketStateCode nextState = SocketStateCode.CLOSED_BY_SERVER; + return to(nextState); + } + + SocketStateChangeResult toUnexpectedClosed() { + SocketStateCode nextState = SocketStateCode.UNEXPECTED_CLOSE_BY_CLIENT; + return to(nextState); + } + + SocketStateChangeResult toUnexpectedClosedByPeer() { + SocketStateCode nextState = SocketStateCode.UNEXPECTED_CLOSE_BY_SERVER; + return to(nextState); + } + + SocketStateChangeResult toErrorUnknown() { + SocketStateCode nextState = SocketStateCode.ERROR_UNKNOWN; + return to(nextState); + } + + private SocketStateChangeResult to(SocketStateCode nextState) { + logger.debug("{} stateTo() started. to:{}", objectUniqName, nextState); + + SocketStateChangeResult stateChangeResult = state.changeState(nextState); + + logger.info("{} stateTo() completed. {}", objectUniqName, stateChangeResult); + + return stateChangeResult; + } + + boolean isBeforeConnected(SocketStateCode currentStateCode) { + return SocketStateCode.isBeforeConnected(currentStateCode); + } + + boolean isEnableCommunication() { + return SocketStateCode.isRun(getCurrentStateCode()); + } + + boolean isEnableCommunication(SocketStateCode currentStateCode) { + return SocketStateCode.isRun(currentStateCode); + } + + boolean isEnableDuplexCommunication() { + return SocketStateCode.isRunDuplex(getCurrentStateCode()); + } + + boolean isClosed() { + return SocketStateCode.isClosed(getCurrentStateCode()); + } + + boolean isClosed(SocketStateCode currentStateCode) { + return SocketStateCode.isClosed(currentStateCode); + } + + boolean onClose(SocketStateCode currentStateCode) { + return SocketStateCode.onClose(currentStateCode); + } + + boolean isReconnect(SocketStateCode currentStateCode) { + if (currentStateCode == SocketStateCode.BEING_CLOSE_BY_SERVER) { + return true; + } + + if (currentStateCode == SocketStateCode.CLOSED_BY_SERVER) { + return true; + } + + if (currentStateCode == SocketStateCode.UNEXPECTED_CLOSE_BY_SERVER) { + return true; + } + + return false; + } + + SocketStateCode getCurrentStateCode() { + return state.getCurrentState(); + } + + @Override + public String toString() { + return state.toString(); + } + +} diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientSocketHandshaker.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandshaker.java similarity index 96% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientSocketHandshaker.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandshaker.java index c6bd351eb..2f3397d51 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientSocketHandshaker.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandshaker.java @@ -1,275 +1,275 @@ -/* - * Copyright 2014 NAVER Corp. - * - * Licensed 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 com.navercorp.pinpoint.rpc.client; - -import java.util.Map; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; - -import org.jboss.netty.channel.Channel; -import org.jboss.netty.channel.ChannelFuture; -import org.jboss.netty.channel.ChannelFutureListener; -import org.jboss.netty.util.Timeout; -import org.jboss.netty.util.Timer; -import org.jboss.netty.util.TimerTask; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.navercorp.pinpoint.rpc.control.ProtocolException; -import com.navercorp.pinpoint.rpc.packet.ControlHandshakePacket; -import com.navercorp.pinpoint.rpc.packet.ControlHandshakeResponsePacket; -import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode; -import com.navercorp.pinpoint.rpc.util.AssertUtils; -import com.navercorp.pinpoint.rpc.util.ClassUtils; -import com.navercorp.pinpoint.rpc.util.ControlMessageEncodingUtils; -import com.navercorp.pinpoint.rpc.util.MapUtils; - -public class PinpointClientSocketHandshaker { - - private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private final ChannelFutureListener handShakeFailFutureListener = new WriteFailFutureListener(this.logger, "HandShakePacket write fail.", "HandShakePacket write success."); - - private static final int STATE_INIT = 0; - private static final int STATE_STARTED = 1; - private static final int STATE_FINISHED = 2; - // STATE_INIT -> STATE_STARTED -> STATE_COMPLETED - // STATE_INIT -> STATE_STARTED -> STATE_ABORTED - private final AtomicInteger state; - - private final AtomicInteger handshakeCount; - - private final Timer handshakerTimer; - private final int retryInterval; - private final int maxHandshakeCount; - - private final Object lock = new Object(); - private final AtomicReference handshakeResult = new AtomicReference(null); - - private String simpleName; - - public PinpointClientSocketHandshaker(Timer handshakerTimer, int retryInterval, int maxHandshakeCount) { - AssertUtils.assertNotNull(handshakerTimer, "handshakerTimer may not be null."); - AssertUtils.assertTrue(retryInterval > 0, "retryInterval must greater than zero."); - AssertUtils.assertTrue(maxHandshakeCount > 0, "maxHandshakeCount must greater than zero."); - - this.state = new AtomicInteger(STATE_INIT); - this.handshakerTimer = handshakerTimer; - this.retryInterval = retryInterval; - this.maxHandshakeCount = maxHandshakeCount; - - this.handshakeCount = new AtomicInteger(0); - } - - public void handshakeStart(Channel channel, Map handshakeData) { - logger.info("{} handshakeStart method started.", simpleClassNameAndHashCodeString()); - - if (channel == null) { - logger.info("{} handshakeStart method failed. channel may not be null.", simpleClassNameAndHashCodeString()); - return; - } - - if (!channel.isConnected()) { - logger.info("{} handshakeStart method failed. channel is not connected.", simpleClassNameAndHashCodeString()); - return; - } - - if (!state.compareAndSet(STATE_INIT, STATE_STARTED)) { - logger.info("{} handshakeStart method failed. currentState:{}", simpleClassNameAndHashCodeString(), state.get()); - return; - } - - HandshakeJob handshakeJob = null; - try { - handshakeJob = createHandshakeJob(channel, handshakeData); - } catch (Exception e) { - if (logger.isWarnEnabled()) { - logger.warn(simpleClassNameAndHashCodeString() + " create handshake job failed. Error:" + e.getMessage() + " state will be aborted.", e); - } - } - - if (handshakeJob == null) { - handshakeAbort(); - logger.info("{} handshakeStart method failed.", simpleClassNameAndHashCodeString()); - return; - } - - handshake(handshakeJob); - reservationJob(handshakeJob); - logger.info("{} handshakeStart method completed. channel:{}, data:{}", simpleClassNameAndHashCodeString(), channel, handshakeData); - } - - private HandshakeJob createHandshakeJob(Channel channel, Map handshakeData) throws ProtocolException { - byte[] payload = ControlMessageEncodingUtils.encode(handshakeData); - ControlHandshakePacket handshakePacket = new ControlHandshakePacket(payload); - - HandshakeJob handshakeJob = new HandshakeJob(channel, handshakePacket); - return handshakeJob; - } - - private void handshake(HandshakeJob handshakeJob) { - handshakeCount.incrementAndGet(); - - Channel channel = handshakeJob.getChannel(); - ControlHandshakePacket packet = handshakeJob.getHandshakePacket(); - - final ChannelFuture future = channel.write(packet); - - logger.debug("{} handshakePacket sent. channel:{}, packet:{}.", simpleClassNameAndHashCodeString(), channel, packet); - - future.addListener(handShakeFailFutureListener); - } - - private void reservationJob(HandshakeJob handshake) { - if (handshakeCount.get() >= maxHandshakeCount) { - handshakeAbort(); - return; - } - - this.handshakerTimer.newTimeout(handshake, retryInterval, TimeUnit.MILLISECONDS); - } - - public boolean handshakeComplete(ControlHandshakeResponsePacket message) { - logger.info("{} handshakeComplete method started. params:{}", simpleClassNameAndHashCodeString(), message); - - synchronized (lock) { - if (!this.state.compareAndSet(STATE_STARTED, STATE_FINISHED)) { - // state can be 0 or 2. - logger.info("{} handshakeComplete method failed. beforeState:{}", simpleClassNameAndHashCodeString(), state.get()); - this.state.set(STATE_FINISHED); - return false; - } - - HandshakeResponseCode code = getHandshakeResponseCode(message); - handshakeResult.compareAndSet(null, code); - logger.info("{} handshakeComplete method completed. handshakeResult:{} / {}", simpleClassNameAndHashCodeString(), code, handshakeResult.get()); - return true; - } - } - - private HandshakeResponseCode getHandshakeResponseCode(ControlHandshakeResponsePacket message) { - byte[] payload = message.getPayload(); - if (payload == null) { - return HandshakeResponseCode.PROTOCOL_ERROR; - } - - try { - Map result = (Map) ControlMessageEncodingUtils.decode(payload); - - int code = MapUtils.getInteger(result, ControlHandshakeResponsePacket.CODE, -1); - int subCode = MapUtils.getInteger(result, ControlHandshakeResponsePacket.SUB_CODE, -1); - - return HandshakeResponseCode.getValue(code, subCode); - } catch (ProtocolException e) { - logger.warn(e.getMessage(), e); - } - - return HandshakeResponseCode.UNKNOWN_CODE; - } - - public HandshakeResponseCode getHandshakeResult() { - return handshakeResult.get(); - } - - public void handshakeAbort() { - logger.info("{} handshakeAbort method started.", simpleClassNameAndHashCodeString()); - - if (!state.compareAndSet(STATE_STARTED, STATE_FINISHED)) { - // state can be 0 or 2. - logger.info("{} handshakeStart method failed. beforeState:{}", simpleClassNameAndHashCodeString(), state.get()); - this.state.set(STATE_FINISHED); - return; - } - logger.info("{} handshakeAbort method completed.", simpleClassNameAndHashCodeString()); - } - - public boolean isRun() { - int currentState = currentState(); - return isRun(currentState); - } - - private boolean isRun(int currentState) { - if (currentState == STATE_STARTED) { - return true; - } else { - return false; - } - } - - public boolean isFinished() { - int currentState = currentState(); - return isFinished(currentState); - } - - private boolean isFinished(int currentState) { - return this.state.get() == STATE_FINISHED; - } - - private int currentState() { - synchronized (lock) { - return this.state.get(); - } - } - - private String simpleClassNameAndHashCodeString() { - if (simpleName == null) { - simpleName = ClassUtils.simpleClassNameAndHashCodeString(this); - } - - return simpleName; - } - - private class HandshakeJob implements TimerTask { - - private final Channel channel; - private final ControlHandshakePacket handshakePacket; - - public HandshakeJob(Channel channel, ControlHandshakePacket handshakePacket) { - this.channel = channel; - this.handshakePacket = handshakePacket; - } - - @Override - public void run(Timeout timeout) throws Exception { - logger.info("Do handshake ({}/{}). channel:{}.", handshakeCount.get(), maxHandshakeCount, channel); - if (timeout.isCancelled()) { - reservationJob(this); - return; - } - - int currentState = currentState(); - - if (isRun(currentState)) { - handshake(this); - reservationJob(this); - } else if (isFinished(currentState)) { - logger.warn("Handshake already completed."); - } else { - logger.warn("Handshake invalid state. {}", state.get()); - } - } - - public Channel getChannel() { - return channel; - } - - public ControlHandshakePacket getHandshakePacket() { - return handshakePacket; - } - } - -} +/* + * Copyright 2014 NAVER Corp. + * + * Licensed 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 com.navercorp.pinpoint.rpc.client; + +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +import org.jboss.netty.channel.Channel; +import org.jboss.netty.channel.ChannelFuture; +import org.jboss.netty.channel.ChannelFutureListener; +import org.jboss.netty.util.Timeout; +import org.jboss.netty.util.Timer; +import org.jboss.netty.util.TimerTask; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.navercorp.pinpoint.rpc.control.ProtocolException; +import com.navercorp.pinpoint.rpc.packet.ControlHandshakePacket; +import com.navercorp.pinpoint.rpc.packet.ControlHandshakeResponsePacket; +import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode; +import com.navercorp.pinpoint.rpc.util.AssertUtils; +import com.navercorp.pinpoint.rpc.util.ClassUtils; +import com.navercorp.pinpoint.rpc.util.ControlMessageEncodingUtils; +import com.navercorp.pinpoint.rpc.util.MapUtils; + +public class PinpointClientHandshaker { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + private final ChannelFutureListener handShakeFailFutureListener = new WriteFailFutureListener(this.logger, "HandShakePacket write fail.", "HandShakePacket write success."); + + private static final int STATE_INIT = 0; + private static final int STATE_STARTED = 1; + private static final int STATE_FINISHED = 2; + // STATE_INIT -> STATE_STARTED -> STATE_COMPLETED + // STATE_INIT -> STATE_STARTED -> STATE_ABORTED + private final AtomicInteger state; + + private final AtomicInteger handshakeCount; + + private final Timer handshakerTimer; + private final int retryInterval; + private final int maxHandshakeCount; + + private final Object lock = new Object(); + private final AtomicReference handshakeResult = new AtomicReference(null); + + private String simpleName; + + public PinpointClientHandshaker(Timer handshakerTimer, int retryInterval, int maxHandshakeCount) { + AssertUtils.assertNotNull(handshakerTimer, "handshakerTimer may not be null."); + AssertUtils.assertTrue(retryInterval > 0, "retryInterval must greater than zero."); + AssertUtils.assertTrue(maxHandshakeCount > 0, "maxHandshakeCount must greater than zero."); + + this.state = new AtomicInteger(STATE_INIT); + this.handshakerTimer = handshakerTimer; + this.retryInterval = retryInterval; + this.maxHandshakeCount = maxHandshakeCount; + + this.handshakeCount = new AtomicInteger(0); + } + + public void handshakeStart(Channel channel, Map handshakeData) { + logger.info("{} handshakeStart method started.", simpleClassNameAndHashCodeString()); + + if (channel == null) { + logger.info("{} handshakeStart method failed. channel may not be null.", simpleClassNameAndHashCodeString()); + return; + } + + if (!channel.isConnected()) { + logger.info("{} handshakeStart method failed. channel is not connected.", simpleClassNameAndHashCodeString()); + return; + } + + if (!state.compareAndSet(STATE_INIT, STATE_STARTED)) { + logger.info("{} handshakeStart method failed. currentState:{}", simpleClassNameAndHashCodeString(), state.get()); + return; + } + + HandshakeJob handshakeJob = null; + try { + handshakeJob = createHandshakeJob(channel, handshakeData); + } catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn(simpleClassNameAndHashCodeString() + " create handshake job failed. Error:" + e.getMessage() + " state will be aborted.", e); + } + } + + if (handshakeJob == null) { + handshakeAbort(); + logger.info("{} handshakeStart method failed.", simpleClassNameAndHashCodeString()); + return; + } + + handshake(handshakeJob); + reservationJob(handshakeJob); + logger.info("{} handshakeStart method completed. channel:{}, data:{}", simpleClassNameAndHashCodeString(), channel, handshakeData); + } + + private HandshakeJob createHandshakeJob(Channel channel, Map handshakeData) throws ProtocolException { + byte[] payload = ControlMessageEncodingUtils.encode(handshakeData); + ControlHandshakePacket handshakePacket = new ControlHandshakePacket(payload); + + HandshakeJob handshakeJob = new HandshakeJob(channel, handshakePacket); + return handshakeJob; + } + + private void handshake(HandshakeJob handshakeJob) { + handshakeCount.incrementAndGet(); + + Channel channel = handshakeJob.getChannel(); + ControlHandshakePacket packet = handshakeJob.getHandshakePacket(); + + final ChannelFuture future = channel.write(packet); + + logger.debug("{} handshakePacket sent. channel:{}, packet:{}.", simpleClassNameAndHashCodeString(), channel, packet); + + future.addListener(handShakeFailFutureListener); + } + + private void reservationJob(HandshakeJob handshake) { + if (handshakeCount.get() >= maxHandshakeCount) { + handshakeAbort(); + return; + } + + this.handshakerTimer.newTimeout(handshake, retryInterval, TimeUnit.MILLISECONDS); + } + + public boolean handshakeComplete(ControlHandshakeResponsePacket message) { + logger.info("{} handshakeComplete method started. params:{}", simpleClassNameAndHashCodeString(), message); + + synchronized (lock) { + if (!this.state.compareAndSet(STATE_STARTED, STATE_FINISHED)) { + // state can be 0 or 2. + logger.info("{} handshakeComplete method failed. beforeState:{}", simpleClassNameAndHashCodeString(), state.get()); + this.state.set(STATE_FINISHED); + return false; + } + + HandshakeResponseCode code = getHandshakeResponseCode(message); + handshakeResult.compareAndSet(null, code); + logger.info("{} handshakeComplete method completed. handshakeResult:{} / {}", simpleClassNameAndHashCodeString(), code, handshakeResult.get()); + return true; + } + } + + private HandshakeResponseCode getHandshakeResponseCode(ControlHandshakeResponsePacket message) { + byte[] payload = message.getPayload(); + if (payload == null) { + return HandshakeResponseCode.PROTOCOL_ERROR; + } + + try { + Map result = (Map) ControlMessageEncodingUtils.decode(payload); + + int code = MapUtils.getInteger(result, ControlHandshakeResponsePacket.CODE, -1); + int subCode = MapUtils.getInteger(result, ControlHandshakeResponsePacket.SUB_CODE, -1); + + return HandshakeResponseCode.getValue(code, subCode); + } catch (ProtocolException e) { + logger.warn(e.getMessage(), e); + } + + return HandshakeResponseCode.UNKNOWN_CODE; + } + + public HandshakeResponseCode getHandshakeResult() { + return handshakeResult.get(); + } + + public void handshakeAbort() { + logger.info("{} handshakeAbort method started.", simpleClassNameAndHashCodeString()); + + if (!state.compareAndSet(STATE_STARTED, STATE_FINISHED)) { + // state can be 0 or 2. + logger.info("{} handshakeStart method failed. beforeState:{}", simpleClassNameAndHashCodeString(), state.get()); + this.state.set(STATE_FINISHED); + return; + } + logger.info("{} handshakeAbort method completed.", simpleClassNameAndHashCodeString()); + } + + public boolean isRun() { + int currentState = currentState(); + return isRun(currentState); + } + + private boolean isRun(int currentState) { + if (currentState == STATE_STARTED) { + return true; + } else { + return false; + } + } + + public boolean isFinished() { + int currentState = currentState(); + return isFinished(currentState); + } + + private boolean isFinished(int currentState) { + return this.state.get() == STATE_FINISHED; + } + + private int currentState() { + synchronized (lock) { + return this.state.get(); + } + } + + private String simpleClassNameAndHashCodeString() { + if (simpleName == null) { + simpleName = ClassUtils.simpleClassNameAndHashCodeString(this); + } + + return simpleName; + } + + private class HandshakeJob implements TimerTask { + + private final Channel channel; + private final ControlHandshakePacket handshakePacket; + + public HandshakeJob(Channel channel, ControlHandshakePacket handshakePacket) { + this.channel = channel; + this.handshakePacket = handshakePacket; + } + + @Override + public void run(Timeout timeout) throws Exception { + logger.info("Do handshake ({}/{}). channel:{}.", handshakeCount.get(), maxHandshakeCount, channel); + if (timeout.isCancelled()) { + reservationJob(this); + return; + } + + int currentState = currentState(); + + if (isRun(currentState)) { + handshake(this); + reservationJob(this); + } else if (isFinished(currentState)) { + logger.warn("Handshake already completed."); + } else { + logger.warn("Handshake invalid state. {}", state.get()); + } + } + + public Channel getChannel() { + return channel; + } + + public ControlHandshakePacket getHandshakePacket() { + return handshakePacket; + } + } + +} diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketClientPipelineFactory.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientPipelineFactory.java similarity index 61% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketClientPipelineFactory.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientPipelineFactory.java index 1ca48a366..cb352dafc 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketClientPipelineFactory.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientPipelineFactory.java @@ -31,15 +31,15 @@ import java.util.concurrent.TimeUnit; * @author emeroad * @author koo.taejin */ -public class SocketClientPipelineFactory implements ChannelPipelineFactory { +public class PinpointClientPipelineFactory implements ChannelPipelineFactory { - private final PinpointSocketFactory pinpointSocketFactory; + private final PinpointClientFactory pinpointClientFactory; - public SocketClientPipelineFactory(PinpointSocketFactory pinpointSocketFactory) { - if (pinpointSocketFactory == null) { - throw new NullPointerException("pinpointSocketFactory must not be null"); + public PinpointClientPipelineFactory(PinpointClientFactory pinpointClientFactory) { + if (pinpointClientFactory == null) { + throw new NullPointerException("pinpointClientFactory must not be null"); } - this.pinpointSocketFactory = pinpointSocketFactory; + this.pinpointClientFactory = pinpointClientFactory; } @@ -49,13 +49,13 @@ public class SocketClientPipelineFactory implements ChannelPipelineFactory { pipeline.addLast("encoder", new PacketEncoder()); pipeline.addLast("decoder", new PacketDecoder()); - long pingDelay = pinpointSocketFactory.getPingDelay(); - long enableWorkerPacketDelay = pinpointSocketFactory.getEnableWorkerPacketDelay(); - long timeoutMillis = pinpointSocketFactory.getTimeoutMillis(); + long pingDelay = pinpointClientFactory.getPingDelay(); + long enableWorkerPacketDelay = pinpointClientFactory.getEnableWorkerPacketDelay(); + long timeoutMillis = pinpointClientFactory.getTimeoutMillis(); - PinpointSocketHandler pinpointSocketHandler = new PinpointSocketHandler(pinpointSocketFactory, pingDelay, enableWorkerPacketDelay, timeoutMillis); - pipeline.addLast("writeTimeout", new WriteTimeoutHandler(pinpointSocketHandler.getChannelTimer(), 3000, TimeUnit.MILLISECONDS)); - pipeline.addLast("socketHandler", pinpointSocketHandler); + DefaultPinpointClientHandler defaultPinpointClientHandler = new DefaultPinpointClientHandler(pinpointClientFactory, pingDelay, enableWorkerPacketDelay, timeoutMillis); + pipeline.addLast("writeTimeout", new WriteTimeoutHandler(defaultPinpointClientHandler.getChannelTimer(), 3000, TimeUnit.MILLISECONDS)); + pipeline.addLast("socketHandler", defaultPinpointClientHandler); return pipeline; } diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketReconnectEventListener.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientReconnectEventListener.java similarity index 88% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketReconnectEventListener.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientReconnectEventListener.java index 62835e8cf..aa3ce166d 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketReconnectEventListener.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientReconnectEventListener.java @@ -16,12 +16,12 @@ package com.navercorp.pinpoint.rpc.client; -public interface PinpointSocketReconnectEventListener { +public interface PinpointClientReconnectEventListener { /* there is no event except "reconnect" currently. when additional events are needed, it will be useful to pass with Event */ - void reconnectPerformed(PinpointSocket socket); + void reconnectPerformed(PinpointClient client); } diff --git a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateSocketHandler.java b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateClientHandler.java similarity index 96% rename from rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateSocketHandler.java rename to rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateClientHandler.java index a2dafe9c3..ce16db48f 100644 --- a/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateSocketHandler.java +++ b/rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateClientHandler.java @@ -32,7 +32,7 @@ import java.net.SocketAddress; * @author emeroad * @author netspider */ -public class ReconnectStateSocketHandler implements SocketHandler { +public class ReconnectStateClientHandler implements PinpointClientHandler { private static final ConnectFuture failedConnectFuture = new ConnectFuture(); static { @@ -56,7 +56,7 @@ public class ReconnectStateSocketHandler implements SocketHandler { } @Override - public void setPinpointSocket(PinpointSocket pinpointSocket) { + public void setPinpointClient(PinpointClient pinpointClient) { } @Override diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ClientMessageListenerTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ClientMessageListenerTest.java index 45bb13057..571e16161 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ClientMessageListenerTest.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ClientMessageListenerTest.java @@ -50,10 +50,10 @@ public class ClientMessageListenerTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new AlwaysHandshakeSuccessListener()); EchoClientListener echoMessageListener = new EchoClientListener(); - PinpointSocketFactory clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), echoMessageListener); + PinpointClientFactory clientSocketFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), echoMessageListener); try { - PinpointSocket socket = clientSocketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientSocketFactory.connect("127.0.0.1", bindPort); Thread.sleep(500); List writableServerList = serverAcceptor.getWritableServerList(); @@ -65,7 +65,7 @@ public class ClientMessageListenerTest { assertSendMessage(writableServer, "simple", echoMessageListener); assertRequestMessage(writableServer, "request", echoMessageListener); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { clientSocketFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); @@ -77,14 +77,14 @@ public class ClientMessageListenerTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new AlwaysHandshakeSuccessListener()); EchoClientListener echoMessageListener1 = PinpointRPCTestUtils.createEchoClientListener(); - PinpointSocketFactory clientSocketFactory1 = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), echoMessageListener1); + PinpointClientFactory clientSocketFactory1 = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), echoMessageListener1); EchoClientListener echoMessageListener2 = PinpointRPCTestUtils.createEchoClientListener(); - PinpointSocketFactory clientSocketFactory2 = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), echoMessageListener2); + PinpointClientFactory clientSocketFactory2 = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), echoMessageListener2); try { - PinpointSocket socket = clientSocketFactory1.connect("127.0.0.1", bindPort); - PinpointSocket socket2 = clientSocketFactory2.connect("127.0.0.1", bindPort); + PinpointClient client = clientSocketFactory1.connect("127.0.0.1", bindPort); + PinpointClient client2 = clientSocketFactory2.connect("127.0.0.1", bindPort); Thread.sleep(500); @@ -102,7 +102,7 @@ public class ClientMessageListenerTest { Assert.assertEquals(1, echoMessageListener1.getRequestPacketRepository().size()); Assert.assertEquals(1, echoMessageListener2.getRequestPacketRepository().size()); - PinpointRPCTestUtils.close(socket, socket2); + PinpointRPCTestUtils.close(client, client2); } finally { clientSocketFactory1.release(); clientSocketFactory2.release(); diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactoryTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactoryTest.java similarity index 75% rename from rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactoryTest.java rename to rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactoryTest.java index 63ab6ed1c..bbe1d417f 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactoryTest.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactoryTest.java @@ -39,31 +39,31 @@ import com.navercorp.pinpoint.rpc.util.PinpointRPCTestUtils; /** * @author emeroad */ -public class PinpointSocketFactoryTest { +public class PinpointClientFactoryTest { private Logger logger = LoggerFactory.getLogger(this.getClass()); private static int bindPort; - private static PinpointSocketFactory socketFactory; + private static PinpointClientFactory clientFactory; @BeforeClass public static void setUp() throws IOException { bindPort = PinpointRPCTestUtils.findAvailablePort(); - socketFactory = new PinpointSocketFactory(); - socketFactory.setPingDelay(100); + clientFactory = new PinpointClientFactory(); + clientFactory.setPingDelay(100); } @AfterClass public static void tearDown() { - if (socketFactory != null) { - socketFactory.release(); + if (clientFactory != null) { + clientFactory.release(); } } @Test public void connectFail() { try { - socketFactory.connect("127.0.0.1", bindPort); + clientFactory.connect("127.0.0.1", bindPort); Assert.fail(); } catch (PinpointSocketException e) { Assert.assertTrue(ConnectException.class.isInstance(e.getCause())); @@ -74,7 +74,7 @@ public class PinpointSocketFactoryTest { public void reconnectFail() throws InterruptedException { // confirm simplified error message when api called. InetSocketAddress remoteAddress = new InetSocketAddress("127.0.0.1", bindPort); - ChannelFuture reconnect = socketFactory.reconnect(remoteAddress); + ChannelFuture reconnect = clientFactory.reconnect(remoteAddress); reconnect.await(); Assert.assertFalse(reconnect.isSuccess()); Assert.assertTrue(ConnectException.class.isInstance(reconnect.getCause())); @@ -87,8 +87,8 @@ public class PinpointSocketFactoryTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort); try { - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); - PinpointRPCTestUtils.close(socket); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); + PinpointRPCTestUtils.close(client); } finally { PinpointRPCTestUtils.close(serverAcceptor); } @@ -99,9 +99,9 @@ public class PinpointSocketFactoryTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort); try { - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); Thread.sleep(1000); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { PinpointRPCTestUtils.close(serverAcceptor); } @@ -112,9 +112,9 @@ public class PinpointSocketFactoryTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort); try { - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); - socket.sendPing(); - PinpointRPCTestUtils.close(socket); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); + client.sendPing(); + PinpointRPCTestUtils.close(client); } finally { PinpointRPCTestUtils.close(serverAcceptor); } @@ -125,13 +125,13 @@ public class PinpointSocketFactoryTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new RequestResponseServerMessageListener()); try { - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); byte[] randomByte = TestByteUtils.createRandomByte(10); - byte[] response = PinpointRPCTestUtils.request(socket, randomByte); + byte[] response = PinpointRPCTestUtils.request(client, randomByte); Assert.assertArrayEquals(randomByte, response); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { PinpointRPCTestUtils.close(serverAcceptor); } @@ -142,13 +142,13 @@ public class PinpointSocketFactoryTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new TestSeverMessageListener()); try { - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); logger.info("send1"); - socket.send(new byte[20]); + client.send(new byte[20]); logger.info("send2"); - socket.sendSync(new byte[20]); + client.sendSync(new byte[20]); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { PinpointRPCTestUtils.close(serverAcceptor); } @@ -159,13 +159,13 @@ public class PinpointSocketFactoryTest { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new TestSeverMessageListener()); try { - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); byte[] randomByte = TestByteUtils.createRandomByte(20); - byte[] response = PinpointRPCTestUtils.request(socket, randomByte); + byte[] response = PinpointRPCTestUtils.request(client, randomByte); Assert.assertArrayEquals(randomByte, response); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { PinpointRPCTestUtils.close(serverAcceptor); } @@ -175,15 +175,15 @@ public class PinpointSocketFactoryTest { public void connectTimeout() { int timeout = 1000; - PinpointSocketFactory pinpointSocketFactory = null; + PinpointClientFactory pinpointClientFactory = null; try { - pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setConnectTimeout(timeout); - int connectTimeout = pinpointSocketFactory.getConnectTimeout(); + pinpointClientFactory = new PinpointClientFactory(); + pinpointClientFactory.setConnectTimeout(timeout); + int connectTimeout = pinpointClientFactory.getConnectTimeout(); Assert.assertEquals(timeout, connectTimeout); } finally { - pinpointSocketFactory.release(); + pinpointClientFactory.release(); } } diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientStateTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientStateTest.java index f6ad62db0..0f2adc614 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientStateTest.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientStateTest.java @@ -47,30 +47,30 @@ public class PinpointClientStateTest { @Test public void connectFailedStateTest() throws InterruptedException { - PinpointSocketFactory clientSocketFactory = null; - PinpointSocketHandler handler = null; + PinpointClientFactory clientFactory = null; + DefaultPinpointClientHandler handler = null; try { - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - handler = connect(clientSocketFactory); + clientFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + handler = connect(clientFactory); Thread.sleep(2000); Assert.assertEquals(SocketStateCode.CONNECT_FAILED, handler.getCurrentStateCode()); } finally { closeHandler(handler); - closeSocketFactory(clientSocketFactory); + closeSocketFactory(clientFactory); } } @Test public void closeStateTest() throws InterruptedException { PinpointServerAcceptor serverAcceptor = null; - PinpointSocketFactory clientSocketFactory = null; - PinpointSocketHandler handler = null; + PinpointClientFactory clientSocketFactory = null; + DefaultPinpointClientHandler handler = null; try { serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, PinpointRPCTestUtils.createEchoServerListener()); - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + clientSocketFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); handler = connect(clientSocketFactory); Thread.sleep(1000); @@ -90,13 +90,13 @@ public class PinpointClientStateTest { @Test public void closeByPeerStateTest() throws InterruptedException { PinpointServerAcceptor serverAcceptor = null; - PinpointSocketFactory clientSocketFactory = null; - PinpointSocketHandler handler = null; + PinpointClientFactory clientFactory = null; + DefaultPinpointClientHandler handler = null; try { serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, PinpointRPCTestUtils.createEchoServerListener()); - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - handler = connect(clientSocketFactory); + clientFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + handler = connect(clientFactory); Thread.sleep(1000); Assert.assertEquals(SocketStateCode.RUN_DUPLEX, handler.getCurrentStateCode()); @@ -107,7 +107,7 @@ public class PinpointClientStateTest { Assert.assertEquals(SocketStateCode.CLOSED_BY_SERVER, handler.getCurrentStateCode()); } finally { closeHandler(handler); - closeSocketFactory(clientSocketFactory); + closeSocketFactory(clientFactory); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -115,24 +115,24 @@ public class PinpointClientStateTest { @Test public void unexpectedCloseStateTest() throws InterruptedException { PinpointServerAcceptor serverAcceptor = null; - PinpointSocketFactory clientSocketFactory = null; - PinpointSocketHandler handler = null; + PinpointClientFactory clientFactory = null; + DefaultPinpointClientHandler handler = null; try { serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, PinpointRPCTestUtils.createEchoServerListener()); - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - handler = connect(clientSocketFactory); + clientFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + handler = connect(clientFactory); Thread.sleep(1000); Assert.assertEquals(SocketStateCode.RUN_DUPLEX, handler.getCurrentStateCode()); - clientSocketFactory.release(); + clientFactory.release(); Thread.sleep(1000); Assert.assertEquals(SocketStateCode.UNEXPECTED_CLOSE_BY_CLIENT, handler.getCurrentStateCode()); } finally { closeHandler(handler); - closeSocketFactory(clientSocketFactory); + closeSocketFactory(clientFactory); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -140,13 +140,13 @@ public class PinpointClientStateTest { @Test public void unexpectedCloseByPeerStateTest() throws InterruptedException { PinpointServerAcceptor serverAcceptor = null; - PinpointSocketFactory clientSocketFactory = null; - PinpointSocketHandler handler = null; + PinpointClientFactory clientFactory = null; + DefaultPinpointClientHandler handler = null; try { serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, PinpointRPCTestUtils.createEchoServerListener()); - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - handler = connect(clientSocketFactory); + clientFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + handler = connect(clientFactory); Thread.sleep(1000); List pinpointServerList = serverAcceptor.getWritableServerList(); @@ -160,36 +160,36 @@ public class PinpointClientStateTest { Assert.assertEquals(SocketStateCode.UNEXPECTED_CLOSE_BY_SERVER, handler.getCurrentStateCode()); } finally { closeHandler(handler); - closeSocketFactory(clientSocketFactory); + closeSocketFactory(clientFactory); PinpointRPCTestUtils.close(serverAcceptor); } } - private PinpointSocketHandler connect(PinpointSocketFactory factory) { + private DefaultPinpointClientHandler connect(PinpointClientFactory factory) { ChannelFuture future = factory.reconnect(new InetSocketAddress("127.0.0.1", bindPort)); - SocketHandler handler = getSocketHandler(future, new InetSocketAddress("127.0.0.1", bindPort)); - return (PinpointSocketHandler) handler; + PinpointClientHandler handler = getSocketHandler(future, new InetSocketAddress("127.0.0.1", bindPort)); + return (DefaultPinpointClientHandler) handler; } - SocketHandler getSocketHandler(ChannelFuture channelConnectFuture, SocketAddress address) { + PinpointClientHandler getSocketHandler(ChannelFuture channelConnectFuture, SocketAddress address) { if (address == null) { throw new NullPointerException("address"); } Channel channel = channelConnectFuture.getChannel(); - SocketHandler socketHandler = (SocketHandler) channel.getPipeline().getLast(); - socketHandler.setConnectSocketAddress(address); + PinpointClientHandler pinpointClientHandler = (PinpointClientHandler) channel.getPipeline().getLast(); + pinpointClientHandler.setConnectSocketAddress(address); - return socketHandler; + return pinpointClientHandler; } - private void closeHandler(PinpointSocketHandler handler) { + private void closeHandler(DefaultPinpointClientHandler handler) { if (handler != null) { handler.close(); } } - private void closeSocketFactory(PinpointSocketFactory factory) { + private void closeSocketFactory(PinpointClientFactory factory) { if (factory != null) { factory.release(); } diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java index 6d799911b..a78850ecc 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java @@ -47,22 +47,22 @@ public class ReconnectTest { private Logger logger = LoggerFactory.getLogger(this.getClass()); private static int bindPort; - private static PinpointSocketFactory socketFactory; + private static PinpointClientFactory clientFactory; @BeforeClass public static void setUp() throws IOException { bindPort = PinpointRPCTestUtils.findAvailablePort(); - socketFactory = new PinpointSocketFactory(); - socketFactory.setReconnectDelay(200); - socketFactory.setPingDelay(100); - socketFactory.setTimeoutMillis(200); + clientFactory = new PinpointClientFactory(); + clientFactory.setReconnectDelay(200); + clientFactory.setPingDelay(100); + clientFactory.setTimeoutMillis(200); } @AfterClass public static void tearDown() { - if (socketFactory != null) { - socketFactory.release(); + if (clientFactory != null) { + clientFactory.release(); } } @@ -75,11 +75,11 @@ public class ReconnectTest { PinpointServerAcceptor newServerAcceptor = null; try { - PinpointSocket socket = socketFactory.connect("localhost", bindPort); - socket.addPinpointSocketReconnectEventListener(new PinpointSocketReconnectEventListener() { + PinpointClient client = clientFactory.connect("localhost", bindPort); + client.addPinpointClientReconnectEventListener(new PinpointClientReconnectEventListener() { @Override - public void reconnectPerformed(PinpointSocket socket) { + public void reconnectPerformed(PinpointClient client) { reconnectPerformed.set(true); } @@ -90,7 +90,7 @@ public class ReconnectTest { logger.info("server.close()---------------------------"); Thread.sleep(1000); try { - byte[] response = PinpointRPCTestUtils.request(socket, new byte[10]); + byte[] response = PinpointRPCTestUtils.request(client, new byte[10]); Assert.fail("expected:exception"); } catch (Exception e) { // skip because of expected error @@ -102,11 +102,11 @@ public class ReconnectTest { Thread.sleep(3000); logger.info("request server---------------------------"); byte[] randomByte = TestByteUtils.createRandomByte(10); - byte[] response = PinpointRPCTestUtils.request(socket, randomByte); + byte[] response = PinpointRPCTestUtils.request(client, randomByte); Assert.assertArrayEquals(randomByte, response); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { PinpointRPCTestUtils.close(newServerAcceptor); } @@ -127,7 +127,7 @@ public class ReconnectTest { logger.info((i + 1) + "th's start."); PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new TestSeverMessageListener()); - PinpointSocket socket = socketFactory.connect("localhost", bindPort); + PinpointClient socket = clientFactory.connect("localhost", bindPort); PinpointRPCTestUtils.close(serverAcceptor); logger.info("server.close()---------------------------"); @@ -155,53 +155,53 @@ public class ReconnectTest { @Test public void scheduledConnect() throws IOException, InterruptedException { - final PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setReconnectDelay(200); - PinpointSocket socket = null; + final PinpointClientFactory clientFactory = new PinpointClientFactory(); + clientFactory.setReconnectDelay(200); + PinpointClient client = null; PinpointServerAcceptor serverAcceptor = null; try { - socket = pinpointSocketFactory.scheduledConnect("localhost", bindPort); + client = clientFactory.scheduledConnect("localhost", bindPort); serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new TestSeverMessageListener()); Thread.sleep(2000); logger.info("request server---------------------------"); byte[] randomByte = TestByteUtils.createRandomByte(10); - byte[] response = PinpointRPCTestUtils.request(socket, randomByte); + byte[] response = PinpointRPCTestUtils.request(client, randomByte); Assert.assertArrayEquals(randomByte, response); } finally { - PinpointRPCTestUtils.close(socket); - pinpointSocketFactory.release(); + PinpointRPCTestUtils.close(client); + clientFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @Test public void scheduledConnectAndClosed() throws IOException, InterruptedException { - PinpointSocket socket = socketFactory.scheduledConnect("localhost", bindPort); + PinpointClient client = clientFactory.scheduledConnect("localhost", bindPort); logger.debug("close"); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } @Test public void scheduledConnectDelayAndClosed() throws IOException, InterruptedException { - PinpointSocket socket = socketFactory.scheduledConnect("localhost", bindPort); + PinpointClient client = clientFactory.scheduledConnect("localhost", bindPort); Thread.sleep(2000); - logger.debug("close pinpoint socket"); - PinpointRPCTestUtils.close(socket); + logger.debug("close pinpoint client"); + PinpointRPCTestUtils.close(client); } @Test public void scheduledConnectStateTest() { - PinpointSocket socket = socketFactory.scheduledConnect("localhost", bindPort); + PinpointClient client = clientFactory.scheduledConnect("localhost", bindPort); - socket.send(new byte[10]); + client.send(new byte[10]); try { - Future future = socket.sendAsync(new byte[10]); + Future future = client.sendAsync(new byte[10]); future.await(); future.getResult(); Assert.fail(); @@ -209,28 +209,28 @@ public class ReconnectTest { } try { - socket.sendSync(new byte[10]); + client.sendSync(new byte[10]); Assert.fail(); } catch (PinpointSocketException e) { } try { - PinpointRPCTestUtils.request(socket, new byte[10]); + PinpointRPCTestUtils.request(client, new byte[10]); Assert.fail(); } catch (PinpointSocketException e) { } - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } @Test public void serverFirstClose() throws IOException, InterruptedException { // when abnormal case in which server has been closed first, confirm that a socket should be closed properly. PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort); - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); byte[] randomByte = TestByteUtils.createRandomByte(10); - Future response = socket.request(randomByte); + Future response = client.request(randomByte); response.await(); try { response.getResult(); @@ -240,7 +240,7 @@ public class ReconnectTest { // close server by force PinpointRPCTestUtils.close(serverAcceptor); Thread.sleep(1000*2); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } @Test @@ -248,13 +248,13 @@ public class ReconnectTest { // when abnormal case in which server has been closed first, confirm that a client socket should be closed properly. PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort); - PinpointSocket socket = socketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); // just close server and request PinpointRPCTestUtils.close(serverAcceptor); byte[] randomByte = TestByteUtils.createRandomByte(10); - Future response = socket.request(randomByte); + Future response = client.request(randomByte); response.await(); try { response.getResult(); @@ -263,7 +263,7 @@ public class ReconnectTest { } Thread.sleep(1000 * 3); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } } diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/HandshakeTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/HandshakeTest.java index d46229db7..c979f7085 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/HandshakeTest.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/HandshakeTest.java @@ -22,6 +22,8 @@ import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.jboss.netty.util.Timer; import org.junit.AfterClass; import org.junit.Assert; @@ -30,9 +32,7 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.navercorp.pinpoint.rpc.client.PinpointClientSocketHandshaker; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; +import com.navercorp.pinpoint.rpc.client.PinpointClientHandshaker; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseType; import com.navercorp.pinpoint.rpc.util.PinpointRPCTestUtils; @@ -64,11 +64,11 @@ public class HandshakeTest { public void handshakeTest1() throws InterruptedException { PinpointServerAcceptor serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, new AlwaysHandshakeSuccessListener()); - PinpointSocketFactory clientSocketFactory1 = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - PinpointSocketFactory clientSocketFactory2 = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), null); + PinpointClientFactory clientFactory1 = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + PinpointClientFactory clientFactory2 = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), null); try { - PinpointSocket socket = clientSocketFactory1.connect("127.0.0.1", bindPort); - PinpointSocket socket2 = clientSocketFactory2.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory1.connect("127.0.0.1", bindPort); + PinpointClient client2 = clientFactory2.connect("127.0.0.1", bindPort); Thread.sleep(500); @@ -77,10 +77,10 @@ public class HandshakeTest { Assert.fail(); } - PinpointRPCTestUtils.close(socket, socket2); + PinpointRPCTestUtils.close(client, client2); } finally { - clientSocketFactory1.release(); - clientSocketFactory2.release(); + clientFactory1.release(); + clientFactory2.release(); PinpointRPCTestUtils.close(serverAcceptor); } @@ -92,10 +92,10 @@ public class HandshakeTest { Map params = PinpointRPCTestUtils.getParams(); - PinpointSocketFactory clientSocketFactory1 = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + PinpointClientFactory clientFactory1 = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); try { - PinpointSocket socket = clientSocketFactory1.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory1.connect("127.0.0.1", bindPort); Thread.sleep(500); PinpointServer writableServer = getWritableServer("application", "agent", (Long) params.get(AgentHandshakePropertyType.START_TIMESTAMP.getName()), serverAcceptor.getWritableServerList()); @@ -104,9 +104,9 @@ public class HandshakeTest { writableServer = getWritableServer("application", "agent", (Long) params.get(AgentHandshakePropertyType.START_TIMESTAMP.getName()) + 1, serverAcceptor.getWritableServerList()); Assert.assertNull(writableServer); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { - clientSocketFactory1.release(); + clientFactory1.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -116,7 +116,7 @@ public class HandshakeTest { int retryInterval = 100; int maxHandshakeCount = 10; - PinpointClientSocketHandshaker handshaker = new PinpointClientSocketHandshaker(timer, retryInterval, maxHandshakeCount); + PinpointClientHandshaker handshaker = new PinpointClientHandshaker(timer, retryInterval, maxHandshakeCount); handshaker.handshakeComplete(null); Assert.assertEquals(null, handshaker.getHandshakeResult()); @@ -129,7 +129,7 @@ public class HandshakeTest { int retryInterval = 100; int maxHandshakeCount = 10; - PinpointClientSocketHandshaker handshaker = new PinpointClientSocketHandshaker(timer, retryInterval, maxHandshakeCount); + PinpointClientHandshaker handshaker = new PinpointClientHandshaker(timer, retryInterval, maxHandshakeCount); handshaker.handshakeAbort(); Assert.assertTrue(handshaker.isFinished()); diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/PinpointServerStateTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/PinpointServerStateTest.java index 6b6c0d38d..ae3e7285e 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/PinpointServerStateTest.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/server/PinpointServerStateTest.java @@ -21,13 +21,13 @@ import java.net.Socket; import java.util.List; import java.util.Map; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.jboss.netty.buffer.ChannelBuffer; import org.junit.Assert; import org.junit.BeforeClass; import org.junit.Test; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; import com.navercorp.pinpoint.rpc.common.SocketStateCode; import com.navercorp.pinpoint.rpc.control.ProtocolException; import com.navercorp.pinpoint.rpc.packet.ControlHandshakePacket; @@ -49,27 +49,27 @@ public class PinpointServerStateTest { @Test public void closeByPeerTest() throws InterruptedException { PinpointServerAcceptor serverAcceptor = null; - PinpointSocket pinpointSocket = null; - PinpointSocketFactory clientSocketFactory = null; + PinpointClient client = null; + PinpointClientFactory clientFactory = null; try { serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, PinpointRPCTestUtils.createEchoServerListener()); - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - pinpointSocket = clientSocketFactory.connect("127.0.0.1", bindPort); + clientFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + client = clientFactory.connect("127.0.0.1", bindPort); Thread.sleep(1000); List pinpointServerList = serverAcceptor.getWritableServerList(); PinpointServer pinpointServer = pinpointServerList.get(0); Assert.assertEquals(SocketStateCode.RUN_DUPLEX, pinpointServer.getCurrentStateCode()); - pinpointSocket.close(); + client.close(); Thread.sleep(1000); Assert.assertEquals(SocketStateCode.CLOSED_BY_CLIENT, pinpointServer.getCurrentStateCode()); } finally { - PinpointRPCTestUtils.close(pinpointSocket); - if (clientSocketFactory != null) { - clientSocketFactory.release(); + PinpointRPCTestUtils.close(client); + if (clientFactory != null) { + clientFactory.release(); } PinpointRPCTestUtils.close(serverAcceptor); } @@ -78,13 +78,13 @@ public class PinpointServerStateTest { @Test public void closeTest() throws InterruptedException { PinpointServerAcceptor serverAcceptor = null; - PinpointSocket pinpointSocket = null; - PinpointSocketFactory clientSocketFactory = null; + PinpointClient client = null; + PinpointClientFactory clientFactory = null; try { serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, PinpointRPCTestUtils.createEchoServerListener()); - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - pinpointSocket = clientSocketFactory.connect("127.0.0.1", bindPort); + clientFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + client = clientFactory.connect("127.0.0.1", bindPort); Thread.sleep(1000); List pinpointServerList = serverAcceptor.getWritableServerList(); @@ -96,9 +96,9 @@ public class PinpointServerStateTest { Assert.assertEquals(SocketStateCode.CLOSED_BY_SERVER, pinpointServer.getCurrentStateCode()); } finally { - PinpointRPCTestUtils.close(pinpointSocket); - if (clientSocketFactory != null) { - clientSocketFactory.release(); + PinpointRPCTestUtils.close(client); + if (clientFactory != null) { + clientFactory.release(); } PinpointRPCTestUtils.close(serverAcceptor); } @@ -131,13 +131,13 @@ public class PinpointServerStateTest { @Test public void unexpectedCloseTest() throws InterruptedException, IOException, ProtocolException { PinpointServerAcceptor serverAcceptor = null; - PinpointSocket pinpointSocket = null; - PinpointSocketFactory clientSocketFactory = null; + PinpointClient client = null; + PinpointClientFactory clientFactory = null; try { serverAcceptor = PinpointRPCTestUtils.createPinpointServerFactory(bindPort, PinpointRPCTestUtils.createEchoServerListener()); - clientSocketFactory = PinpointRPCTestUtils.createSocketFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); - pinpointSocket = clientSocketFactory.connect("127.0.0.1", bindPort); + clientFactory = PinpointRPCTestUtils.createClientFactory(PinpointRPCTestUtils.getParams(), PinpointRPCTestUtils.createEchoClientListener()); + client = clientFactory.connect("127.0.0.1", bindPort); Thread.sleep(1000); List pinpointServerList = serverAcceptor.getWritableServerList(); @@ -149,9 +149,9 @@ public class PinpointServerStateTest { Assert.assertEquals(SocketStateCode.UNEXPECTED_CLOSE_BY_SERVER, pinpointServer.getCurrentStateCode()); } finally { - PinpointRPCTestUtils.close(pinpointSocket); - if (clientSocketFactory != null) { - clientSocketFactory.release(); + PinpointRPCTestUtils.close(client); + if (clientFactory != null) { + clientFactory.release(); } PinpointRPCTestUtils.close(serverAcceptor); } diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/stream/StreamChannelManagerTest.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/stream/StreamChannelManagerTest.java index 82527f2b3..dba6ab84c 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/stream/StreamChannelManagerTest.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/stream/StreamChannelManagerTest.java @@ -20,6 +20,8 @@ import java.io.IOException; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.junit.Assert; import org.junit.BeforeClass; @@ -29,8 +31,6 @@ import com.navercorp.pinpoint.rpc.PinpointSocketException; import com.navercorp.pinpoint.rpc.RecordedStreamChannelMessageListener; import com.navercorp.pinpoint.rpc.TestByteUtils; import com.navercorp.pinpoint.rpc.client.MessageListener; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; import com.navercorp.pinpoint.rpc.client.SimpleLoggingMessageListener; import com.navercorp.pinpoint.rpc.packet.stream.StreamClosePacket; import com.navercorp.pinpoint.rpc.packet.stream.StreamCreatePacket; @@ -57,13 +57,13 @@ public class StreamChannelManagerTest { PinpointServerAcceptor serverAcceptor = createServerFactory(new TestSeverMessageListener(), new ServerListener(bo)); serverAcceptor.bind("localhost", bindPort); - PinpointSocketFactory pinpointSocketFactory = createSocketFactory(); + PinpointClientFactory clientFactory = createSocketFactory(); try { - PinpointSocket socket = pinpointSocketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); RecordedStreamChannelMessageListener clientListener = new RecordedStreamChannelMessageListener(4); - ClientStreamChannelContext clientContext = socket.createStreamChannel(new byte[0], clientListener); + ClientStreamChannelContext clientContext = client.createStreamChannel(new byte[0], clientListener); int sendCount = 4; @@ -77,9 +77,9 @@ public class StreamChannelManagerTest { clientContext.getStreamChannel().close(); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { - pinpointSocketFactory.release(); + clientFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -92,15 +92,15 @@ public class StreamChannelManagerTest { PinpointServerAcceptor serverAcceptor = createServerFactory(new TestSeverMessageListener(), new ServerListener(bo)); serverAcceptor.bind("localhost", bindPort); - PinpointSocketFactory pinpointSocketFactory = createSocketFactory(); + PinpointClientFactory clientFactory = createSocketFactory(); try { - PinpointSocket socket = pinpointSocketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); RecordedStreamChannelMessageListener clientListener = new RecordedStreamChannelMessageListener(4); - ClientStreamChannelContext clientContext = socket.createStreamChannel(new byte[0], clientListener); + ClientStreamChannelContext clientContext = client.createStreamChannel(new byte[0], clientListener); RecordedStreamChannelMessageListener clientListener2 = new RecordedStreamChannelMessageListener(4); - ClientStreamChannelContext clientContext2 = socket.createStreamChannel(new byte[0], clientListener2); + ClientStreamChannelContext clientContext2 = client.createStreamChannel(new byte[0], clientListener2); int sendCount = 4; @@ -130,9 +130,9 @@ public class StreamChannelManagerTest { clientContext2.getStreamChannel().close(); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { - pinpointSocketFactory.release(); + clientFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -144,10 +144,10 @@ public class StreamChannelManagerTest { SimpleStreamBO bo = new SimpleStreamBO(); - PinpointSocketFactory pinpointSocketFactory = createSocketFactory(new TestListener(), new ServerListener(bo)); + PinpointClientFactory clientFactory = createSocketFactory(new TestListener(), new ServerListener(bo)); try { - PinpointSocket socket = pinpointSocketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); Thread.sleep(100); @@ -172,9 +172,9 @@ public class StreamChannelManagerTest { clientContext.getStreamChannel().close(); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { - pinpointSocketFactory.release(); + clientFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -184,21 +184,21 @@ public class StreamChannelManagerTest { PinpointServerAcceptor serverAcceptor = createServerFactory(new TestSeverMessageListener(), null); serverAcceptor.bind("localhost", bindPort); - PinpointSocketFactory pinpointSocketFactory = createSocketFactory(); + PinpointClientFactory clientFactory = createSocketFactory(); try { - PinpointSocket socket = pinpointSocketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); RecordedStreamChannelMessageListener clientListener = new RecordedStreamChannelMessageListener(4); - ClientStreamChannelContext clientContext = socket.createStreamChannel(new byte[0], clientListener); + ClientStreamChannelContext clientContext = client.createStreamChannel(new byte[0], clientListener); Thread.sleep(100); clientContext.getStreamChannel().close(); - PinpointRPCTestUtils.close(socket); + PinpointRPCTestUtils.close(client); } finally { - pinpointSocketFactory.release(); + clientFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -210,15 +210,15 @@ public class StreamChannelManagerTest { PinpointServerAcceptor serverAcceptor = createServerFactory(new TestSeverMessageListener(), new ServerListener(bo)); serverAcceptor.bind("localhost", bindPort); - PinpointSocketFactory pinpointSocketFactory = createSocketFactory(); + PinpointClientFactory clientFactory = createSocketFactory(); - PinpointSocket socket = null; + PinpointClient client = null; try { - socket = pinpointSocketFactory.connect("127.0.0.1", bindPort); + client = clientFactory.connect("127.0.0.1", bindPort); RecordedStreamChannelMessageListener clientListener = new RecordedStreamChannelMessageListener(4); - ClientStreamChannelContext clientContext = socket.createStreamChannel(new byte[0], clientListener); + ClientStreamChannelContext clientContext = client.createStreamChannel(new byte[0], clientListener); Thread.sleep(100); Assert.assertEquals(1, bo.getStreamChannelContextSize()); @@ -229,8 +229,8 @@ public class StreamChannelManagerTest { Assert.assertEquals(0, bo.getStreamChannelContextSize()); } finally { - PinpointRPCTestUtils.close(socket); - pinpointSocketFactory.release(); + PinpointRPCTestUtils.close(client); + clientFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -246,9 +246,9 @@ public class StreamChannelManagerTest { SimpleStreamBO bo = new SimpleStreamBO(); - PinpointSocketFactory pinpointSocketFactory = createSocketFactory(new TestListener(), new ServerListener(bo)); + PinpointClientFactory clientFactory = createSocketFactory(new TestListener(), new ServerListener(bo)); - PinpointSocket socket = pinpointSocketFactory.connect("127.0.0.1", bindPort); + PinpointClient client = clientFactory.connect("127.0.0.1", bindPort); try { Thread.sleep(100); @@ -263,7 +263,7 @@ public class StreamChannelManagerTest { ClientStreamChannelContext clientContext = writableServer.createStream(new byte[0], clientListener); - StreamChannelContext aaa = socket.findStreamChannel(2); + StreamChannelContext aaa = client.findStreamChannel(2); aaa.getStreamChannel().close(); @@ -274,8 +274,8 @@ public class StreamChannelManagerTest { clientContext.getStreamChannel().close(); } finally { - PinpointRPCTestUtils.close(socket); - pinpointSocketFactory.release(); + PinpointRPCTestUtils.close(client); + clientFactory.release(); PinpointRPCTestUtils.close(serverAcceptor); } } @@ -295,17 +295,17 @@ public class StreamChannelManagerTest { return serverAcceptor; } - private PinpointSocketFactory createSocketFactory() { - PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - return pinpointSocketFactory; + private PinpointClientFactory createSocketFactory() { + PinpointClientFactory clientFactory = new PinpointClientFactory(); + return clientFactory; } - private PinpointSocketFactory createSocketFactory(MessageListener messageListener, ServerStreamChannelMessageListener serverStreamChannelMessageListener) { - PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); - pinpointSocketFactory.setMessageListener(messageListener); - pinpointSocketFactory.setServerStreamChannelMessageListener(serverStreamChannelMessageListener); + private PinpointClientFactory createSocketFactory(MessageListener messageListener, ServerStreamChannelMessageListener serverStreamChannelMessageListener) { + PinpointClientFactory clientFactory = new PinpointClientFactory(); + clientFactory.setMessageListener(messageListener); + clientFactory.setServerStreamChannelMessageListener(serverStreamChannelMessageListener); - return pinpointSocketFactory; + return clientFactory; } class TestListener extends SimpleLoggingMessageListener { diff --git a/rpc/src/test/java/com/navercorp/pinpoint/rpc/util/PinpointRPCTestUtils.java b/rpc/src/test/java/com/navercorp/pinpoint/rpc/util/PinpointRPCTestUtils.java index dcd4659f7..21a5833fe 100644 --- a/rpc/src/test/java/com/navercorp/pinpoint/rpc/util/PinpointRPCTestUtils.java +++ b/rpc/src/test/java/com/navercorp/pinpoint/rpc/util/PinpointRPCTestUtils.java @@ -24,6 +24,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import com.navercorp.pinpoint.rpc.client.PinpointClient; import org.jboss.netty.channel.Channel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,8 +32,7 @@ import org.slf4j.LoggerFactory; import com.navercorp.pinpoint.rpc.Future; import com.navercorp.pinpoint.rpc.ResponseMessage; import com.navercorp.pinpoint.rpc.client.MessageListener; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode; import com.navercorp.pinpoint.rpc.packet.HandshakeResponseType; import com.navercorp.pinpoint.rpc.packet.PingPacket; @@ -105,12 +105,12 @@ public final class PinpointRPCTestUtils { } } - public static PinpointSocketFactory createSocketFactory(Map param) { - return createSocketFactory(param, null); + public static PinpointClientFactory createClientFactory(Map param) { + return createClientFactory(param, null); } - public static PinpointSocketFactory createSocketFactory(Map param, MessageListener messageListener) { - PinpointSocketFactory socketFactory = new PinpointSocketFactory(); + public static PinpointClientFactory createClientFactory(Map param, MessageListener messageListener) { + PinpointClientFactory socketFactory = new PinpointClientFactory(); socketFactory.setProperties(param); if (messageListener != null) { @@ -126,19 +126,19 @@ public final class PinpointRPCTestUtils { return future.getResult().getMessage(); } - public static byte[] request(PinpointSocket pinpointSocket, byte[] message) { - Future future = pinpointSocket.request(message); + public static byte[] request(PinpointClient client, byte[] message) { + Future future = client.request(message); future.await(); return future.getResult().getMessage(); } - public static void close(PinpointSocket socket, PinpointSocket... sockets) { - if (socket != null) { - socket.close(); + public static void close(PinpointClient client, PinpointClient... clients) { + if (client != null) { + client.close(); } - if (sockets != null) { - for (PinpointSocket eachSocket : sockets) { + if (clients != null) { + for (PinpointClient eachSocket : clients) { if (eachSocket != null) { eachSocket.close(); } diff --git a/web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java b/web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java index 1ebbc4434..2fcbb8016 100644 --- a/web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java +++ b/web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java @@ -22,6 +22,8 @@ import static org.mockito.Mockito.when; import java.io.IOException; import java.util.List; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import org.junit.Assert; import org.apache.curator.test.TestingServer; @@ -37,8 +39,6 @@ import org.slf4j.LoggerFactory; import com.navercorp.pinpoint.common.util.NetUtils; import com.navercorp.pinpoint.rpc.client.MessageListener; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; import com.navercorp.pinpoint.rpc.packet.RequestPacket; import com.navercorp.pinpoint.rpc.packet.SendPacket; import com.navercorp.pinpoint.web.config.WebConfig; @@ -149,8 +149,8 @@ public class ClusterTest { public void clusterTest3() throws Exception { ts.restart(); - PinpointSocketFactory factory = null; - PinpointSocket socket = null; + PinpointClientFactory clientFactory = null; + PinpointClient client = null; ZooKeeper zookeeper = null; try { @@ -161,17 +161,17 @@ public class ClusterTest { Assert.assertEquals(0, socketManager.getCollectorList().size()); - factory = new PinpointSocketFactory(); - factory.setMessageListener(new SimpleListener()); + clientFactory = new PinpointClientFactory(); + clientFactory.setMessageListener(new SimpleListener()); - socket = factory.connect(DEFAULT_IP, acceptorPort); + client = clientFactory.connect(DEFAULT_IP, acceptorPort); Thread.sleep(1000); Assert.assertEquals(1, socketManager.getCollectorList().size()); } finally { - closePinpointSocket(factory, socket); + closePinpointSocket(clientFactory, client); if (zookeeper != null) { zookeeper.close(); @@ -212,13 +212,13 @@ public class ClusterTest { } } - private void closePinpointSocket(PinpointSocketFactory factory, PinpointSocket socket) { - if (socket != null) { - socket.close(); + private void closePinpointSocket(PinpointClientFactory clientFactory, PinpointClient client) { + if (client != null) { + client.close(); } - if (factory != null) { - factory.release(); + if (clientFactory != null) { + clientFactory.release(); } } diff --git a/web/src/test/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterTest.java b/web/src/test/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterTest.java index 048fb52d3..8975d5f5c 100644 --- a/web/src/test/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterTest.java +++ b/web/src/test/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterTest.java @@ -37,8 +37,8 @@ import org.slf4j.LoggerFactory; import com.navercorp.pinpoint.collector.cluster.zookeeper.exception.PinpointZookeeperException; import com.navercorp.pinpoint.common.util.NetUtils; import com.navercorp.pinpoint.rpc.client.MessageListener; -import com.navercorp.pinpoint.rpc.client.PinpointSocket; -import com.navercorp.pinpoint.rpc.client.PinpointSocketFactory; +import com.navercorp.pinpoint.rpc.client.PinpointClient; +import com.navercorp.pinpoint.rpc.client.PinpointClientFactory; import com.navercorp.pinpoint.rpc.packet.RequestPacket; import com.navercorp.pinpoint.rpc.packet.SendPacket; import com.navercorp.pinpoint.web.util.PinpointWebTestUtils; @@ -199,13 +199,13 @@ public class ZookeeperClusterTest { } } - private void closePinpointSocket(PinpointSocketFactory factory, PinpointSocket socket) { - if (socket != null) { - socket.close(); + private void closeResources(PinpointClientFactory clientFactory, PinpointClient client) { + if (client != null) { + client.close(); } - if (factory != null) { - factory.release(); + if (clientFactory != null) { + clientFactory.release(); } }