You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by hx...@apache.org on 2020/04/16 14:01:08 UTC
[incubator-iotdb] branch disable_mqtt_server updated: fix sync
server startup problem in UTs
This is an automated email from the ASF dual-hosted git repository.
hxd pushed a commit to branch disable_mqtt_server
in repository https://gitbox.apache.org/repos/asf/incubator-iotdb.git
The following commit(s) were added to refs/heads/disable_mqtt_server by this push:
new ca14146 fix sync server startup problem in UTs
ca14146 is described below
commit ca14146bfe048597ffe163ca36dcf31a1d2bbb23
Author: xiangdong huang <sa...@gmail.com>
AuthorDate: Thu Apr 16 22:00:50 2020 +0800
fix sync server startup problem in UTs
---
.../org/apache/iotdb/db/service/RPCService.java | 15 +-
.../iotdb/db/sync/receiver/SyncServerManager.java | 28 +-
.../db/sync/thrift/SyncServiceEventHandler.java | 56 ----
.../thrift/server/CustomizedTThreadPoolServer.java | 367 ---------------------
.../thrift/transport/CustomizedTServerSocket.java | 138 --------
5 files changed, 22 insertions(+), 582 deletions(-)
diff --git a/server/src/main/java/org/apache/iotdb/db/service/RPCService.java b/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
index 31ae0a6..52b4b79 100644
--- a/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
+++ b/server/src/main/java/org/apache/iotdb/db/service/RPCService.java
@@ -32,10 +32,9 @@ import org.apache.iotdb.service.rpc.thrift.TSIService.Processor;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
-import org.apache.thrift.server.CustomizedTThreadPoolServer;
import org.apache.thrift.server.TServer;
-import org.apache.thrift.server.CustomizedTThreadPoolServer.Args;
-import org.apache.thrift.transport.CustomizedTServerSocket;
+import org.apache.thrift.server.TThreadPoolServer;
+import org.apache.thrift.transport.TServerSocket;
import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -54,7 +53,7 @@ public class RPCService implements RPCServiceMBean, IService {
private RPCServiceThread rpcServiceThread;
private TProtocolFactory protocolFactory;
private Processor<TSIService.Iface> processor;
- private CustomizedTThreadPoolServer.Args poolArgs;
+ private TThreadPoolServer.Args poolArgs;
private TSServiceImpl impl;
private CountDownLatch stopLatch;
@@ -179,7 +178,7 @@ public class RPCService implements RPCServiceMBean, IService {
private class RPCServiceThread extends Thread {
- private CustomizedTServerSocket serverTransport;
+ private TServerSocket serverTransport;
private TServer poolServer;
private CountDownLatch threadStopLatch;
@@ -203,13 +202,13 @@ public class RPCService implements RPCServiceMBean, IService {
logger.info("The RPC service thread begin to run...");
try {
IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig();
- serverTransport = new CustomizedTServerSocket(new InetSocketAddress(config.getRpcAddress(),
+ serverTransport = new TServerSocket(new InetSocketAddress(config.getRpcAddress(),
config.getRpcPort()));
//this is for testing.
if (!serverTransport.getServerSocket().isBound()) {
logger.error("The RPC service port is not bound.");
}
- poolArgs = new Args(serverTransport).maxWorkerThreads(IoTDBDescriptor.
+ poolArgs = new TThreadPoolServer.Args(serverTransport).maxWorkerThreads(IoTDBDescriptor.
getInstance().getConfig().getRpcMaxConcurrentClientNum()).minWorkerThreads(1)
.stopTimeoutVal(
IoTDBDescriptor.getInstance().getConfig().getThriftServerAwaitTimeForStopService());
@@ -217,7 +216,7 @@ public class RPCService implements RPCServiceMBean, IService {
ThreadName.RPC_CLIENT.getName());
poolArgs.processor(processor);
poolArgs.protocolFactory(protocolFactory);
- poolServer = new CustomizedTThreadPoolServer(poolArgs);
+ poolServer = new TThreadPoolServer(poolArgs);
poolServer.serve();
} catch (TTransportException e) {
throw new RPCServiceException(String.format("%s: failed to start %s, because ", IoTDBConstant.GLOBAL_DB_NAME,
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/receiver/SyncServerManager.java b/server/src/main/java/org/apache/iotdb/db/sync/receiver/SyncServerManager.java
index 19c4d51..f1fd2f1 100644
--- a/server/src/main/java/org/apache/iotdb/db/sync/receiver/SyncServerManager.java
+++ b/server/src/main/java/org/apache/iotdb/db/sync/receiver/SyncServerManager.java
@@ -32,7 +32,6 @@ import org.apache.iotdb.db.service.ServiceType;
import org.apache.iotdb.db.sync.receiver.load.FileLoaderManager;
import org.apache.iotdb.db.sync.receiver.recover.SyncReceiverLogAnalyzer;
import org.apache.iotdb.db.sync.receiver.transfer.SyncServiceImpl;
-import org.apache.iotdb.db.sync.thrift.SyncServiceEventHandler;
import org.apache.iotdb.service.sync.thrift.SyncService;
import org.apache.iotdb.service.sync.thrift.SyncService.Processor;
import org.apache.thrift.protocol.TBinaryProtocol;
@@ -54,10 +53,8 @@ public class SyncServerManager implements IService {
private IoTDBConfig conf = IoTDBDescriptor.getInstance().getConfig();
- private Thread syncServerThread;
+ private SyncServiceThread syncServerThread;
- //we add this latch for avoiding in some ITs, the syncService is not startup but the IT has finished.
- private CountDownLatch startLatch;
//stopLatch is also for letting the IT know whether the socket is closed.
private CountDownLatch stopLatch;
@@ -87,14 +84,16 @@ public class SyncServerManager implements IService {
"Sync server failed to start because IP white list is null, please set IP white list.");
return;
}
- startLatch = new CountDownLatch(1);
stopLatch = new CountDownLatch(1);
conf.setIpWhiteList(conf.getIpWhiteList().replaceAll(" ", ""));
- syncServerThread = new SyncServiceThread(startLatch, stopLatch);
+ syncServerThread = new SyncServiceThread(stopLatch);
syncServerThread.setName(ThreadName.SYNC_SERVER.getName());
syncServerThread.start();
try {
- startLatch.await();
+ while (!syncServerThread.isServing()) {
+ //sleep 100ms for waiting the sync server start.
+ Thread.sleep(100);
+ }
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new StartupException(this.getID().getName(), e.getMessage());
@@ -109,7 +108,7 @@ public class SyncServerManager implements IService {
public void stop() {
if (conf.isSyncEnable()) {
FileLoaderManager.getInstance().stop();
- ((SyncServiceThread) syncServerThread).close();
+ syncServerThread.close();
try {
stopLatch.await();
} catch (InterruptedException e) {
@@ -136,13 +135,10 @@ public class SyncServerManager implements IService {
private TProtocolFactory protocolFactory;
private Processor<SyncService.Iface> processor;
private TThreadPoolServer.Args poolArgs;
- //we add this latch for avoiding in some ITs, the syncService is not startup but the IT has finished.
- private CountDownLatch threadStartLatch;
private CountDownLatch threadStopLatch;
- public SyncServiceThread(CountDownLatch startLatch, CountDownLatch stopLatch) {
+ public SyncServiceThread(CountDownLatch stopLatch) {
processor = new SyncService.Processor<>(new SyncServiceImpl());
- this.threadStartLatch = startLatch;
this.threadStopLatch = stopLatch;
}
@@ -164,7 +160,6 @@ public class SyncServerManager implements IService {
poolArgs.protocolFactory(protocolFactory);
poolArgs.processor(processor);
poolServer = new TThreadPoolServer(poolArgs);
- poolServer.setServerEventHandler(new SyncServiceEventHandler(threadStartLatch));
poolServer.serve();
} catch (TTransportException e) {
logger.error("{}: failed to start {}, because ", IoTDBConstant.GLOBAL_DB_NAME,
@@ -192,5 +187,12 @@ public class SyncServerManager implements IService {
serverTransport = null;
}
}
+
+ boolean isServing() {
+ if (poolServer != null) {
+ return poolServer.isServing();
+ }
+ return false;
+ }
}
}
\ No newline at end of file
diff --git a/server/src/main/java/org/apache/iotdb/db/sync/thrift/SyncServiceEventHandler.java b/server/src/main/java/org/apache/iotdb/db/sync/thrift/SyncServiceEventHandler.java
deleted file mode 100644
index 25469ab..0000000
--- a/server/src/main/java/org/apache/iotdb/db/sync/thrift/SyncServiceEventHandler.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.db.sync.thrift;
-
-import java.util.concurrent.CountDownLatch;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.server.ServerContext;
-import org.apache.thrift.server.TServerEventHandler;
-import org.apache.thrift.transport.TTransport;
-
-public class SyncServiceEventHandler implements TServerEventHandler {
-
- private CountDownLatch startLatch;
-
- public SyncServiceEventHandler(CountDownLatch startLatch) {
- this.startLatch = startLatch;
- }
-
- @Override
- public void preServe() {
- startLatch.countDown();
- }
-
- @Override
- public ServerContext createContext(TProtocol input, TProtocol output) {
- return null;
- }
-
- @Override
- public void deleteContext(ServerContext serverContext, TProtocol input, TProtocol output) {
-
- }
-
- @Override
- public void processContext(ServerContext serverContext, TTransport inputTransport,
- TTransport outputTransport) {
-
- }
-}
diff --git a/server/src/main/java/org/apache/thrift/server/CustomizedTThreadPoolServer.java b/server/src/main/java/org/apache/thrift/server/CustomizedTThreadPoolServer.java
deleted file mode 100644
index 37ad757..0000000
--- a/server/src/main/java/org/apache/thrift/server/CustomizedTThreadPoolServer.java
+++ /dev/null
@@ -1,367 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.thrift.server;
-
-import java.util.Random;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.SynchronousQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import java.util.concurrent.TimeUnit;
-
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.thrift.TException;
-import org.apache.thrift.TProcessor;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TServerTransport;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Server which uses Java's built in ThreadPool management to spawn off
- * a worker pool that
- *
- * Because there may be bug in TThreadPoolServer, we have to copy it here and see what is wrong.
- */
-public class CustomizedTThreadPoolServer extends TServer {
- private static final Logger LOGGER = LoggerFactory.getLogger(CustomizedTThreadPoolServer.class);
-
- private volatile AtomicBoolean stoppedCheck = new AtomicBoolean(true);
-
- public static class Args extends AbstractServerArgs<Args> {
- public int minWorkerThreads = 5;
- public int maxWorkerThreads = Integer.MAX_VALUE;
- public ExecutorService executorService;
- public int stopTimeoutVal = 60;
- public TimeUnit stopTimeoutUnit = TimeUnit.SECONDS;
- public int requestTimeout = 20;
- public TimeUnit requestTimeoutUnit = TimeUnit.SECONDS;
- public int beBackoffSlotLength = 100;
- public TimeUnit beBackoffSlotLengthUnit = TimeUnit.MILLISECONDS;
-
- public Args(TServerTransport transport) {
- super(transport);
- }
-
- public Args minWorkerThreads(int n) {
- minWorkerThreads = n;
- return this;
- }
-
- public Args maxWorkerThreads(int n) {
- maxWorkerThreads = n;
- return this;
- }
-
- public Args stopTimeoutVal(int n) {
- stopTimeoutVal = n;
- return this;
- }
-
- public Args stopTimeoutUnit(TimeUnit tu) {
- stopTimeoutUnit = tu;
- return this;
- }
-
- public Args requestTimeout(int n) {
- requestTimeout = n;
- return this;
- }
-
- public Args requestTimeoutUnit(TimeUnit tu) {
- requestTimeoutUnit = tu;
- return this;
- }
- //Binary exponential backoff slot length
- public Args beBackoffSlotLength(int n) {
- beBackoffSlotLength = n;
- return this;
- }
-
- //Binary exponential backoff slot time unit
- public Args beBackoffSlotLengthUnit(TimeUnit tu) {
- beBackoffSlotLengthUnit = tu;
- return this;
- }
-
- public Args executorService(ExecutorService executorService) {
- this.executorService = executorService;
- return this;
- }
- }
-
- // Executor service for handling client connections
- private ExecutorService executorService_;
-
- private final TimeUnit stopTimeoutUnit;
-
- private final long stopTimeoutVal;
-
- private final TimeUnit requestTimeoutUnit;
-
- private final long requestTimeout;
-
- private final long beBackoffSlotInMillis;
-
- private Random random = new Random(System.currentTimeMillis());
-
- public CustomizedTThreadPoolServer(Args args) {
- super(args);
-
- stopTimeoutUnit = args.stopTimeoutUnit;
- stopTimeoutVal = args.stopTimeoutVal;
- requestTimeoutUnit = args.requestTimeoutUnit;
- requestTimeout = args.requestTimeout;
- beBackoffSlotInMillis = args.beBackoffSlotLengthUnit.toMillis(args.beBackoffSlotLength);
-
- executorService_ = args.executorService != null ?
- args.executorService : createDefaultExecutorService(args);
- }
-
- private static ExecutorService createDefaultExecutorService(Args args) {
- SynchronousQueue<Runnable> executorQueue =
- new SynchronousQueue<Runnable>();
- return new ThreadPoolExecutor(args.minWorkerThreads,
- args.maxWorkerThreads,
- args.stopTimeoutVal,
- args.stopTimeoutUnit,
- executorQueue);
- }
-
- protected ExecutorService getExecutorService() {
- return executorService_;
- }
-
- protected boolean preServe() {
- try {
- serverTransport_.listen();
- } catch (TTransportException ttx) {
- LOGGER.error("Error occurred during listening.", ttx);
- return false;
- }
-
- // Run the preServe event
- if (eventHandler_ != null) {
- eventHandler_.preServe();
- }
- LOGGER.info("set stopped_ to false. Memory instance {}", this);
- stoppedCheck.set(false);
- stopped_ = false;
- setServing(true);
-
- return true;
- }
-
- public void serve() {
- if (!preServe()) {
- return;
- }
-
- execute();
- waitForShutdown();
-
- setServing(false);
- }
-
- protected void execute() {
- int failureCount = 0;
- while (!this.stopped_) {
- try {
- TTransport client = serverTransport_.accept();
- WorkerProcess wp = new WorkerProcess(client);
-
- int retryCount = 0;
- long remainTimeInMillis = requestTimeoutUnit.toMillis(requestTimeout);
- while(true) {
- try {
- executorService_.execute(wp);
- break;
- } catch(Throwable t) {
- if (t instanceof RejectedExecutionException) {
- retryCount++;
- try {
- if (remainTimeInMillis > 0) {
- //do a truncated 20 binary exponential backoff sleep
- long sleepTimeInMillis = ((long) (random.nextDouble() *
- (1L << Math.min(retryCount, 20)))) * beBackoffSlotInMillis;
- sleepTimeInMillis = Math.min(sleepTimeInMillis, remainTimeInMillis);
- TimeUnit.MILLISECONDS.sleep(sleepTimeInMillis);
- remainTimeInMillis = remainTimeInMillis - sleepTimeInMillis;
- } else {
- client.close();
- wp = null;
- LOGGER.warn("Task has been rejected by ExecutorService " + retryCount
- + " times till timedout, reason: " + t);
- break;
- }
- } catch (InterruptedException e) {
- LOGGER.warn("Interrupted while waiting to place client on executor queue.");
- Thread.currentThread().interrupt();
- break;
- }
- } else if (t instanceof Error) {
- LOGGER.error("ExecutorService threw error: " + t, t);
- throw (Error)t;
- } else {
- //for other possible runtime errors from ExecutorService, should also not kill serve
- LOGGER.warn("ExecutorService threw error: " + t, t);
- break;
- }
- }
- }
- } catch (TTransportException ttx) {
- if (!stopped_) {
- ++failureCount;
- LOGGER.warn("Stopped: {}. Transport error occurred during acceptance of message. Memory instance {}", stopped_, this, ttx);
- stopped_ = stoppedCheck.get();
- LOGGER.info("update Stopped to {}.", stopped_);
- }
- }
- }
- }
-
- protected void waitForShutdown() {
- executorService_.shutdown();
-
- // Loop until awaitTermination finally does return without a interrupted
- // exception. If we don't do this, then we'll shut down prematurely. We want
- // to let the executorService clear it's task queue, closing client sockets
- // appropriately.
- long timeoutMS = stopTimeoutUnit.toMillis(stopTimeoutVal);
- long now = System.currentTimeMillis();
- while (timeoutMS >= 0) {
- try {
- executorService_.awaitTermination(timeoutMS, TimeUnit.MILLISECONDS);
- break;
- } catch (InterruptedException ix) {
- long newnow = System.currentTimeMillis();
- timeoutMS -= (newnow - now);
- now = newnow;
- }
- }
- }
-
- public void stop() {
- LOGGER.info("set stopped_ to true. Memory instance {}", this);
- stoppedCheck.set(true);
- super.stopped_ = true;
- serverTransport_.interrupt();
- }
-
- private class WorkerProcess implements Runnable {
-
- /**
- * Client that this services.
- */
- private TTransport client_;
-
- /**
- * Default constructor.
- *
- * @param client Transport to process
- */
- private WorkerProcess(TTransport client) {
- client_ = client;
- }
-
- /**
- * Loops on processing a client forever
- */
- public void run() {
- TProcessor processor = null;
- TTransport inputTransport = null;
- TTransport outputTransport = null;
- TProtocol inputProtocol = null;
- TProtocol outputProtocol = null;
-
- TServerEventHandler eventHandler = null;
- ServerContext connectionContext = null;
-
- try {
- processor = processorFactory_.getProcessor(client_);
- inputTransport = inputTransportFactory_.getTransport(client_);
- outputTransport = outputTransportFactory_.getTransport(client_);
- inputProtocol = inputProtocolFactory_.getProtocol(inputTransport);
- outputProtocol = outputProtocolFactory_.getProtocol(outputTransport);
-
- eventHandler = getEventHandler();
- if (eventHandler != null) {
- connectionContext = eventHandler.createContext(inputProtocol, outputProtocol);
- }
- // we check stopped_ first to make sure we're not supposed to be shutting
- // down. this is necessary for graceful shutdown.
- while (true) {
-
- if (eventHandler != null) {
- eventHandler.processContext(connectionContext, inputTransport, outputTransport);
- }
-
- if (stopped_) {
- break;
- }
- processor.process(inputProtocol, outputProtocol);
- }
- } catch (Exception x) {
- // We'll usually receive RuntimeException types here
- // Need to unwrap to ascertain real causing exception before we choose to ignore
- // Ignore err-logging all transport-level/type exceptions
- if (!isIgnorableException(x)) {
- // Log the exception at error level and continue
- LOGGER.error((x instanceof TException? "Thrift " : "") + "Error occurred during processing of message.", x);
- }
- } finally {
- if (eventHandler != null) {
- eventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
- }
- if (inputTransport != null) {
- inputTransport.close();
- }
- if (outputTransport != null) {
- outputTransport.close();
- }
- if (client_.isOpen()) {
- client_.close();
- }
- }
- }
-
- private boolean isIgnorableException(Exception x) {
- TTransportException tTransportException = null;
-
- if (x instanceof TTransportException) {
- tTransportException = (TTransportException)x;
- }
- else if (x.getCause() instanceof TTransportException) {
- tTransportException = (TTransportException)x.getCause();
- }
-
- if (tTransportException != null) {
- switch(tTransportException.getType()) {
- case TTransportException.END_OF_FILE:
- case TTransportException.TIMED_OUT:
- return true;
- }
- }
- return false;
- }
- }
-}
diff --git a/server/src/main/java/org/apache/thrift/transport/CustomizedTServerSocket.java b/server/src/main/java/org/apache/thrift/transport/CustomizedTServerSocket.java
deleted file mode 100644
index 05714fd..0000000
--- a/server/src/main/java/org/apache/thrift/transport/CustomizedTServerSocket.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.thrift.transport;
-
-import java.net.SocketOptions;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.SocketException;
-
-/**
- * Wrapper around ServerSocket for Thrift.
- *
- */
-public class CustomizedTServerSocket extends TServerTransport {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(CustomizedTServerSocket.class);
-
- /**
- * Underlying ServerSocket object
- */
- private ServerSocket serverSocket_ = null;
-
- /**
- * Timeout for client sockets from accept
- */
- private int clientTimeout_ = 0;
-
- public static class ServerSocketTransportArgs extends AbstractServerTransportArgs<ServerSocketTransportArgs> {
- ServerSocket serverSocket;
-
- public ServerSocketTransportArgs serverSocket(ServerSocket serverSocket) {
- this.serverSocket = serverSocket;
- return this;
- }
-
- public ServerSocketTransportArgs bindAddr(InetSocketAddress bindAddr) {
- super.bindAddr(bindAddr);
- return this;
- }
- }
-
- public CustomizedTServerSocket(InetSocketAddress bindAddr) throws TTransportException {
- this(bindAddr, 0);
- }
-
- public CustomizedTServerSocket(InetSocketAddress bindAddr, int clientTimeout) throws TTransportException {
- this(new ServerSocketTransportArgs().bindAddr(bindAddr).clientTimeout(clientTimeout));
- }
-
- public CustomizedTServerSocket(ServerSocketTransportArgs args) throws TTransportException {
- clientTimeout_ = args.clientTimeout;
- if (args.serverSocket != null) {
- LOGGER.info("server socket is not null, reuse it.");
- this.serverSocket_ = args.serverSocket;
- return;
- }
- try {
- // Make server socket
- serverSocket_ = new ServerSocket();
- // Prevent 2MSL delay problem on server restarts
- serverSocket_.setReuseAddress(true);
- // Bind to listening port
- serverSocket_.bind(args.bindAddr, args.backlog);
- LOGGER.info("server socket is null, generate one {}.", args.bindAddr);
- } catch (IOException ioe) {
- close();
- throw new TTransportException("Could not create ServerSocket on address " + args.bindAddr.toString() + ".", ioe);
- }
- }
-
- public void listen() throws TTransportException {
- // Make sure to block on accept
- if (serverSocket_ != null) {
- try {
- serverSocket_.setSoTimeout(0);
- } catch (SocketException sx) {
- LOGGER.error("Could not set socket timeout.", sx);
- }
- }
- }
-
- protected TSocket acceptImpl() throws TTransportException {
- if (serverSocket_ == null) {
- throw new TTransportException(TTransportException.NOT_OPEN, "No underlying server socket.");
- }
- try {
- Socket result = serverSocket_.accept();
- TSocket result2 = new TSocket(result);
- result2.setTimeout(clientTimeout_);
- return result2;
- } catch (IOException iox) {
- throw new TTransportException(iox);
- }
- }
-
- public void close() {
- if (serverSocket_ != null) {
- try {
- serverSocket_.close();
- } catch (IOException iox) {
- LOGGER.warn("Could not close server socket.", iox);
- }
- serverSocket_ = null;
- }
- }
-
- public void interrupt() {
- // The thread-safeness of this is dubious, but Java documentation suggests
- // that it is safe to do this from a different thread context
- close();
- }
-
- public ServerSocket getServerSocket() {
- return serverSocket_;
- }
-}