home/pinpoint
2
0
Fork 0
mirror of https://github.com/wahyd4/pinpoint.git synced 2026-08-16 08:16:15 +10:00
Code Issues Packages Projects Releases Wiki Activity

rename some classes in rpc project. #989

Browse Source 
socket -> client
socket (use generic concept for client and server)
This commit is contained in:
koo.taejin
2015-09-22 17:40:31 +09:00
parent 6b911fca54
commit 51578e9501
34 changed files with 1053 additions and 1061 deletions
Download Patch File Download Diff File Expand all files Collapse all files
collector/src/main/java/com/navercorp/pinpoint/collector/cluster/WebCluster.java
+23 -23
View File
@@ -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<InetSocketAddress, PinpointSocket> clusterRepository = new HashMap<InetSocketAddress, PinpointSocket>();
private final Map<InetSocketAddress, PinpointClient> clusterRepository = new HashMap<InetSocketAddress, PinpointClient>();
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<String, Object> properties = new HashMap<String, Object>();
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<InetSocketAddress> 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();
}
}
profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java
+24 -25
View File
@@ -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<String, Object> 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();
}
}
profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/EnhancedDataSender.java
+3 -3
View File
@@ -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<ResponseMessage> listener);
boolean addReconnectEventListener(PinpointSocketReconnectEventListener eventListener);
boolean removeReconnectEventListener(PinpointSocketReconnectEventListener eventListener);
boolean addReconnectEventListener(PinpointClientReconnectEventListener eventListener);
boolean removeReconnectEventListener(PinpointClientReconnectEventListener eventListener);
}
profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/LoggingDataSender.java
+3 -3
View File
@@ -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;
}
profiler/src/main/java/com/navercorp/pinpoint/profiler/sender/TcpDataSender.java
+13 -13
View File
@@ -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<Object> 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<ResponseMessage> response = this.socket.request(requestPacket);
final Future<ResponseMessage> 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();
}
}
profiler/src/main/java/com/navercorp/pinpoint/profiler/tools/NetworkAvailabilityChecker.java
+19 -20
View File
@@ -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.<String, Object>emptyMap());
private static PinpointClientFactory createPinpointClientFactory() {
PinpointClientFactory pinpointClientFactory = new PinpointClientFactory();
pinpointClientFactory.setTimeoutMillis(1000 * 5);
pinpointClientFactory.setProperties(Collections.<String, Object>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;
profiler/src/main/java/com/navercorp/pinpoint/test/TestTcpDataSender.java
+3 -3
View File
@@ -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;
}
profiler/src/test/java/com/navercorp/pinpoint/profiler/AgentInfoSenderTest.java
+44 -44
View File
@@ -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<ServerMetaData> serverMetaDataObjects = new ArrayList<ServerMetaData>();
serverMetaDataObjects.add(new DefaultServerMetaData("server1", Collections.<String>emptyList(), Collections.<Integer, String>emptyMap(), Collections.<ServiceInfo>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.<String>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.<String, Object>emptyMap());
private PinpointClientFactory createPinpointClientFactory() {
PinpointClientFactory clientFactory = new PinpointClientFactory();
clientFactory.setTimeoutMillis(1000 * 5);
clientFactory.setProperties(Collections.<String, Object>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;
}
}
profiler/src/test/java/com/navercorp/pinpoint/profiler/context/TraceTest.java
+3 -8
View File
@@ -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;
}
profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/CountingDataSender.java
+3 -4
View File
@@ -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;
}
profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderReconnectTest.java
+18 -18
View File
@@ -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;
}
}
profiler/src/test/java/com/navercorp/pinpoint/profiler/sender/TcpDataSenderTest.java
+20 -20
View File
@@ -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;
}
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/ClassPreLoader.java
+11 -11
View File
@@ -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();
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandler.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DefaultPinpointClientHandler.java
+33 -33
View File
@@ -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<String, Object> handshakeData = new HashMap<String, Object>();
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
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointSocketReconnectEventListener.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/DummyPinpointClientReconnectEventListener.java
+2 -2
View File
@@ -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) {
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocket.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClient.java
+35 -35
View File
@@ -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<PinpointSocketReconnectEventListener> reconnectEventListeners = new CopyOnWriteArrayList<PinpointSocketReconnectEventListener>();
private List<PinpointClientReconnectEventListener> reconnectEventListeners = new CopyOnWriteArrayList<PinpointClientReconnectEventListener>();
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<ResponseMessage> 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<ResponseMessage> returnFailureFuture() {
DefaultFuture<ResponseMessage> future = new DefaultFuture<ResponseMessage>();
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();
}
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactory.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactory.java
+46 -46
View File
@@ -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.");
}
}
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketHandler.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandler.java
+2 -2
View File
@@ -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);
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerContext.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerContext.java
+65 -65
View File
@@ -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);
}
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketHandlerState.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandlerState.java
+169 -169
View File
@@ -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();
}
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientSocketHandshaker.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientHandshaker.java
+275 -275
View File
@@ -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<HandshakeResponseCode> handshakeResult = new AtomicReference<HandshakeResponseCode>(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<String, Object> 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<String, Object> 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<HandshakeResponseCode> handshakeResult = new AtomicReference<HandshakeResponseCode>(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<String, Object> 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<String, Object> 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;
}
}
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/SocketClientPipelineFactory.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientPipelineFactory.java
+12 -12
View File
@@ -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;
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointSocketReconnectEventListener.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/PinpointClientReconnectEventListener.java
+2 -2
View File
@@ -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);
}
rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateSocketHandler.java → rpc/src/main/java/com/navercorp/pinpoint/rpc/client/ReconnectStateClientHandler.java
+2 -2
View File
@@ -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
rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ClientMessageListenerTest.java
+8 -8
View File
@@ -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<PinpointServer> 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();
rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointSocketFactoryTest.java → rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientFactoryTest.java
+30 -30
View File
@@ -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();
}
}
rpc/src/test/java/com/navercorp/pinpoint/rpc/client/PinpointClientStateTest.java
+33 -33
View File
@@ -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<PinpointServer> 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();
}
rpc/src/test/java/com/navercorp/pinpoint/rpc/client/ReconnectTest.java
+38 -38
View File
@@ -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<ResponseMessage> response = socket.request(randomByte);
Future<ResponseMessage> 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<ResponseMessage> response = socket.request(randomByte);
Future<ResponseMessage> 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);
}
}
rpc/src/test/java/com/navercorp/pinpoint/rpc/server/HandshakeTest.java
+16 -16
View File
@@ -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());
rpc/src/test/java/com/navercorp/pinpoint/rpc/server/PinpointServerStateTest.java
+24 -24
View File
@@ -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<PinpointServer> 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<PinpointServer> 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<PinpointServer> 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);
}
rpc/src/test/java/com/navercorp/pinpoint/rpc/stream/StreamChannelManagerTest.java
+41 -41
View File
@@ -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 {
rpc/src/test/java/com/navercorp/pinpoint/rpc/util/PinpointRPCTestUtils.java
+13 -13
View File
@@ -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<ResponseMessage> future = pinpointSocket.request(message);
public static byte[] request(PinpointClient client, byte[] message) {
Future<ResponseMessage> 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();
}
web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java
+13 -13
View File
@@ -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();
}
}
web/src/test/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterTest.java
+7 -7
View File
@@ -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();
}
}