001/** 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.activemq.transport.tcp; 018 019import java.io.IOException; 020import java.net.InetAddress; 021import java.net.InetSocketAddress; 022import java.net.ServerSocket; 023import java.net.Socket; 024import java.net.SocketException; 025import java.net.SocketTimeoutException; 026import java.net.URI; 027import java.net.URISyntaxException; 028import java.net.UnknownHostException; 029import java.nio.channels.SelectionKey; 030import java.nio.channels.ServerSocketChannel; 031import java.nio.channels.SocketChannel; 032import java.util.HashMap; 033import java.util.concurrent.BlockingQueue; 034import java.util.concurrent.LinkedBlockingQueue; 035import java.util.concurrent.TimeUnit; 036import java.util.concurrent.atomic.AtomicInteger; 037 038import javax.net.ServerSocketFactory; 039import javax.net.ssl.SSLServerSocket; 040 041import org.apache.activemq.Service; 042import org.apache.activemq.ThreadPriorities; 043import org.apache.activemq.TransportLoggerSupport; 044import org.apache.activemq.command.BrokerInfo; 045import org.apache.activemq.openwire.OpenWireFormatFactory; 046import org.apache.activemq.transport.Transport; 047import org.apache.activemq.transport.TransportServer; 048import org.apache.activemq.transport.TransportServerThreadSupport; 049import org.apache.activemq.transport.nio.SelectorManager; 050import org.apache.activemq.transport.nio.SelectorSelection; 051import org.apache.activemq.util.IOExceptionSupport; 052import org.apache.activemq.util.InetAddressUtil; 053import org.apache.activemq.util.IntrospectionSupport; 054import org.apache.activemq.util.ServiceListener; 055import org.apache.activemq.util.ServiceStopper; 056import org.apache.activemq.util.ServiceSupport; 057import org.apache.activemq.wireformat.WireFormat; 058import org.apache.activemq.wireformat.WireFormatFactory; 059import org.slf4j.Logger; 060import org.slf4j.LoggerFactory; 061 062/** 063 * A TCP based implementation of {@link TransportServer} 064 */ 065public class TcpTransportServer extends TransportServerThreadSupport implements ServiceListener { 066 067 private static final Logger LOG = LoggerFactory.getLogger(TcpTransportServer.class); 068 protected ServerSocket serverSocket; 069 protected SelectorSelection selector; 070 protected int backlog = 5000; 071 protected WireFormatFactory wireFormatFactory = new OpenWireFormatFactory(); 072 protected final TcpTransportFactory transportFactory; 073 protected long maxInactivityDuration = 30000; 074 protected long maxInactivityDurationInitalDelay = 10000; 075 protected int minmumWireFormatVersion; 076 protected boolean useQueueForAccept = true; 077 protected boolean allowLinkStealing; 078 079 /** 080 * trace=true -> the Transport stack where this TcpTransport object will be, will have a TransportLogger layer 081 * trace=false -> the Transport stack where this TcpTransport object will be, will NOT have a TransportLogger layer, 082 * and therefore will never be able to print logging messages. This parameter is most probably set in Connection or 083 * TransportConnector URIs. 084 */ 085 protected boolean trace = false; 086 087 protected int soTimeout = 0; 088 protected int socketBufferSize = 64 * 1024; 089 protected int connectionTimeout = 30000; 090 091 /** 092 * Name of the LogWriter implementation to use. Names are mapped to classes in the 093 * resources/META-INF/services/org/apache/activemq/transport/logwriters directory. This parameter is most probably 094 * set in Connection or TransportConnector URIs. 095 */ 096 protected String logWriterName = TransportLoggerSupport.defaultLogWriterName; 097 098 /** 099 * Specifies if the TransportLogger will be manageable by JMX or not. Also, as long as there is at least 1 100 * TransportLogger which is manageable, a TransportLoggerControl MBean will me created. 101 */ 102 protected boolean dynamicManagement = false; 103 104 /** 105 * startLogging=true -> the TransportLogger object of the Transport stack will initially write messages to the log. 106 * startLogging=false -> the TransportLogger object of the Transport stack will initially NOT write messages to the 107 * log. This parameter only has an effect if trace == true. This parameter is most probably set in Connection or 108 * TransportConnector URIs. 109 */ 110 protected boolean startLogging = true; 111 protected final ServerSocketFactory serverSocketFactory; 112 protected BlockingQueue<Socket> socketQueue = new LinkedBlockingQueue<Socket>(); 113 protected Thread socketHandlerThread; 114 115 /** 116 * The maximum number of sockets allowed for this server 117 */ 118 protected int maximumConnections = Integer.MAX_VALUE; 119 protected AtomicInteger currentTransportCount = new AtomicInteger(); 120 121 public TcpTransportServer(TcpTransportFactory transportFactory, URI location, ServerSocketFactory serverSocketFactory) throws IOException, 122 URISyntaxException { 123 super(location); 124 this.transportFactory = transportFactory; 125 this.serverSocketFactory = serverSocketFactory; 126 } 127 128 public void bind() throws IOException { 129 URI bind = getBindLocation(); 130 131 String host = bind.getHost(); 132 host = (host == null || host.length() == 0) ? "localhost" : host; 133 InetAddress addr = InetAddress.getByName(host); 134 135 try { 136 this.serverSocket = serverSocketFactory.createServerSocket(bind.getPort(), backlog, addr); 137 configureServerSocket(this.serverSocket); 138 } catch (IOException e) { 139 throw IOExceptionSupport.create("Failed to bind to server socket: " + bind + " due to: " + e, e); 140 } 141 try { 142 setConnectURI(new URI(bind.getScheme(), bind.getUserInfo(), resolveHostName(serverSocket, addr), serverSocket.getLocalPort(), bind.getPath(), 143 bind.getQuery(), bind.getFragment())); 144 } catch (URISyntaxException e) { 145 146 // it could be that the host name contains invalid characters such 147 // as _ on unix platforms so lets try use the IP address instead 148 try { 149 setConnectURI(new URI(bind.getScheme(), bind.getUserInfo(), addr.getHostAddress(), serverSocket.getLocalPort(), bind.getPath(), 150 bind.getQuery(), bind.getFragment())); 151 } catch (URISyntaxException e2) { 152 throw IOExceptionSupport.create(e2); 153 } 154 } 155 } 156 157 private void configureServerSocket(ServerSocket socket) throws SocketException { 158 socket.setSoTimeout(2000); 159 if (transportOptions != null) { 160 161 // If the enabledCipherSuites option is invalid we don't want to ignore it as the call 162 // to SSLServerSocket to configure it has a side effect on the socket rendering it 163 // useless as all suites are enabled many of which are considered as insecure. We 164 // instead trap that option here and throw an exception. We should really consider 165 // all invalid options as breaking and not start the transport but the current design 166 // doesn't really allow for this. 167 // 168 // see: https://issues.apache.org/jira/browse/AMQ-4582 169 // 170 if (socket instanceof SSLServerSocket) { 171 if (transportOptions.containsKey("enabledCipherSuites")) { 172 Object cipherSuites = transportOptions.remove("enabledCipherSuites"); 173 174 if (!IntrospectionSupport.setProperty(socket, "enabledCipherSuites", cipherSuites)) { 175 throw new SocketException(String.format( 176 "Invalid transport options {enabledCipherSuites=%s}", cipherSuites)); 177 } 178 } 179 } 180 181 IntrospectionSupport.setProperties(socket, transportOptions); 182 } 183 } 184 185 /** 186 * @return Returns the wireFormatFactory. 187 */ 188 public WireFormatFactory getWireFormatFactory() { 189 return wireFormatFactory; 190 } 191 192 /** 193 * @param wireFormatFactory 194 * The wireFormatFactory to set. 195 */ 196 public void setWireFormatFactory(WireFormatFactory wireFormatFactory) { 197 this.wireFormatFactory = wireFormatFactory; 198 } 199 200 /** 201 * Associates a broker info with the transport server so that the transport can do discovery advertisements of the 202 * broker. 203 * 204 * @param brokerInfo 205 */ 206 @Override 207 public void setBrokerInfo(BrokerInfo brokerInfo) { 208 } 209 210 public long getMaxInactivityDuration() { 211 return maxInactivityDuration; 212 } 213 214 public void setMaxInactivityDuration(long maxInactivityDuration) { 215 this.maxInactivityDuration = maxInactivityDuration; 216 } 217 218 public long getMaxInactivityDurationInitalDelay() { 219 return this.maxInactivityDurationInitalDelay; 220 } 221 222 public void setMaxInactivityDurationInitalDelay(long maxInactivityDurationInitalDelay) { 223 this.maxInactivityDurationInitalDelay = maxInactivityDurationInitalDelay; 224 } 225 226 public int getMinmumWireFormatVersion() { 227 return minmumWireFormatVersion; 228 } 229 230 public void setMinmumWireFormatVersion(int minmumWireFormatVersion) { 231 this.minmumWireFormatVersion = minmumWireFormatVersion; 232 } 233 234 public boolean isTrace() { 235 return trace; 236 } 237 238 public void setTrace(boolean trace) { 239 this.trace = trace; 240 } 241 242 public String getLogWriterName() { 243 return logWriterName; 244 } 245 246 public void setLogWriterName(String logFormat) { 247 this.logWriterName = logFormat; 248 } 249 250 public boolean isDynamicManagement() { 251 return dynamicManagement; 252 } 253 254 public void setDynamicManagement(boolean useJmx) { 255 this.dynamicManagement = useJmx; 256 } 257 258 public boolean isStartLogging() { 259 return startLogging; 260 } 261 262 public void setStartLogging(boolean startLogging) { 263 this.startLogging = startLogging; 264 } 265 266 /** 267 * @return the backlog 268 */ 269 public int getBacklog() { 270 return backlog; 271 } 272 273 /** 274 * @param backlog 275 * the backlog to set 276 */ 277 public void setBacklog(int backlog) { 278 this.backlog = backlog; 279 } 280 281 /** 282 * @return the useQueueForAccept 283 */ 284 public boolean isUseQueueForAccept() { 285 return useQueueForAccept; 286 } 287 288 /** 289 * @param useQueueForAccept 290 * the useQueueForAccept to set 291 */ 292 public void setUseQueueForAccept(boolean useQueueForAccept) { 293 this.useQueueForAccept = useQueueForAccept; 294 } 295 296 /** 297 * pull Sockets from the ServerSocket 298 */ 299 @Override 300 public void run() { 301 final ServerSocketChannel chan = serverSocket.getChannel(); 302 if (chan != null) { 303 try { 304 chan.configureBlocking(false); 305 selector = SelectorManager.getInstance().register(chan, new SelectorManager.Listener() { 306 @Override 307 public void onSelect(SelectorSelection sel) { 308 try { 309 SocketChannel sc = chan.accept(); 310 if (sc != null) { 311 if (isStopped() || getAcceptListener() == null) { 312 sc.close(); 313 } else { 314 if (useQueueForAccept) { 315 socketQueue.put(sc.socket()); 316 } else { 317 handleSocket(sc.socket()); 318 } 319 } 320 } 321 } catch (Exception e) { 322 onError(sel, e); 323 } 324 } 325 @Override 326 public void onError(SelectorSelection sel, Throwable error) { 327 Exception e = null; 328 if (error instanceof Exception) { 329 e = (Exception)error; 330 } else { 331 e = new Exception(error); 332 } 333 if (!isStopping()) { 334 onAcceptError(e); 335 } else if (!isStopped()) { 336 LOG.warn("run()", e); 337 onAcceptError(e); 338 } 339 } 340 }); 341 selector.setInterestOps(SelectionKey.OP_ACCEPT); 342 selector.enable(); 343 } catch (IOException ex) { 344 selector = null; 345 } 346 } else { 347 while (!isStopped()) { 348 Socket socket = null; 349 try { 350 socket = serverSocket.accept(); 351 if (socket != null) { 352 if (isStopped() || getAcceptListener() == null) { 353 socket.close(); 354 } else { 355 if (useQueueForAccept) { 356 socketQueue.put(socket); 357 } else { 358 handleSocket(socket); 359 } 360 } 361 } 362 } catch (SocketTimeoutException ste) { 363 // expect this to happen 364 } catch (Exception e) { 365 if (!isStopping()) { 366 onAcceptError(e); 367 } else if (!isStopped()) { 368 LOG.warn("run()", e); 369 onAcceptError(e); 370 } 371 } 372 } 373 } 374 } 375 376 /** 377 * Allow derived classes to override the Transport implementation that this transport server creates. 378 * 379 * @param socket 380 * @param format 381 * @return 382 * @throws IOException 383 */ 384 protected Transport createTransport(Socket socket, WireFormat format) throws IOException { 385 return new TcpTransport(format, socket); 386 } 387 388 /** 389 * @return pretty print of this 390 */ 391 @Override 392 public String toString() { 393 return "" + getBindLocation(); 394 } 395 396 /** 397 * @param socket 398 * @param bindAddress 399 * @return real hostName 400 * @throws UnknownHostException 401 */ 402 protected String resolveHostName(ServerSocket socket, InetAddress bindAddress) throws UnknownHostException { 403 String result = null; 404 if (socket.isBound()) { 405 if (socket.getInetAddress().isAnyLocalAddress()) { 406 // make it more human readable and useful, an alternative to 0.0.0.0 407 result = InetAddressUtil.getLocalHostName(); 408 } else { 409 result = socket.getInetAddress().getCanonicalHostName(); 410 } 411 } else { 412 result = bindAddress.getCanonicalHostName(); 413 } 414 return result; 415 } 416 417 @Override 418 protected void doStart() throws Exception { 419 if (useQueueForAccept) { 420 Runnable run = new Runnable() { 421 @Override 422 public void run() { 423 try { 424 while (!isStopped() && !isStopping()) { 425 Socket sock = socketQueue.poll(1, TimeUnit.SECONDS); 426 if (sock != null) { 427 try { 428 handleSocket(sock); 429 } catch (Throwable thrown) { 430 if (!isStopping()) { 431 onAcceptError(new Exception(thrown)); 432 } else if (!isStopped()) { 433 LOG.warn("Unexpected error thrown during accept handling: ", thrown); 434 onAcceptError(new Exception(thrown)); 435 } 436 } 437 } 438 } 439 440 } catch (InterruptedException e) { 441 LOG.info("socketQueue interuppted - stopping"); 442 if (!isStopping()) { 443 onAcceptError(e); 444 } 445 } 446 } 447 }; 448 socketHandlerThread = new Thread(null, run, "ActiveMQ Transport Server Thread Handler: " + toString(), getStackSize()); 449 socketHandlerThread.setDaemon(true); 450 socketHandlerThread.setPriority(ThreadPriorities.BROKER_MANAGEMENT - 1); 451 socketHandlerThread.start(); 452 } 453 super.doStart(); 454 } 455 456 @Override 457 protected void doStop(ServiceStopper stopper) throws Exception { 458 if (selector != null) { 459 selector.disable(); 460 selector.close(); 461 selector = null; 462 } 463 if (serverSocket != null) { 464 serverSocket.close(); 465 serverSocket = null; 466 } 467 super.doStop(stopper); 468 } 469 470 @Override 471 public InetSocketAddress getSocketAddress() { 472 return (InetSocketAddress) serverSocket.getLocalSocketAddress(); 473 } 474 475 protected final void handleSocket(Socket socket) { 476 boolean closeSocket = true; 477 try { 478 if (this.currentTransportCount.get() >= this.maximumConnections) { 479 throw new ExceededMaximumConnectionsException( 480 "Exceeded the maximum number of allowed client connections. See the '" + 481 "maximumConnections' property on the TCP transport configuration URI " + 482 "in the ActiveMQ configuration file (e.g., activemq.xml)"); 483 } else { 484 HashMap<String, Object> options = new HashMap<String, Object>(); 485 options.put("maxInactivityDuration", Long.valueOf(maxInactivityDuration)); 486 options.put("maxInactivityDurationInitalDelay", Long.valueOf(maxInactivityDurationInitalDelay)); 487 options.put("minmumWireFormatVersion", Integer.valueOf(minmumWireFormatVersion)); 488 options.put("trace", Boolean.valueOf(trace)); 489 options.put("soTimeout", Integer.valueOf(soTimeout)); 490 options.put("socketBufferSize", Integer.valueOf(socketBufferSize)); 491 options.put("connectionTimeout", Integer.valueOf(connectionTimeout)); 492 options.put("logWriterName", logWriterName); 493 options.put("dynamicManagement", Boolean.valueOf(dynamicManagement)); 494 options.put("startLogging", Boolean.valueOf(startLogging)); 495 options.putAll(transportOptions); 496 497 WireFormat format = wireFormatFactory.createWireFormat(); 498 Transport transport = createTransport(socket, format); 499 closeSocket = false; 500 501 if (transport instanceof ServiceSupport) { 502 ((ServiceSupport) transport).addServiceListener(this); 503 } 504 505 Transport configuredTransport = transportFactory.serverConfigure(transport, format, options); 506 507 getAcceptListener().onAccept(configuredTransport); 508 currentTransportCount.incrementAndGet(); 509 } 510 } catch (SocketTimeoutException ste) { 511 // expect this to happen 512 } catch (Exception e) { 513 if (closeSocket) { 514 try { 515 socket.close(); 516 } catch (Exception ignore) { 517 } 518 } 519 520 if (!isStopping()) { 521 onAcceptError(e); 522 } else if (!isStopped()) { 523 LOG.warn("run()", e); 524 onAcceptError(e); 525 } 526 } 527 } 528 529 public int getSoTimeout() { 530 return soTimeout; 531 } 532 533 public void setSoTimeout(int soTimeout) { 534 this.soTimeout = soTimeout; 535 } 536 537 public int getSocketBufferSize() { 538 return socketBufferSize; 539 } 540 541 public void setSocketBufferSize(int socketBufferSize) { 542 this.socketBufferSize = socketBufferSize; 543 } 544 545 public int getConnectionTimeout() { 546 return connectionTimeout; 547 } 548 549 public void setConnectionTimeout(int connectionTimeout) { 550 this.connectionTimeout = connectionTimeout; 551 } 552 553 /** 554 * @return the maximumConnections 555 */ 556 public int getMaximumConnections() { 557 return maximumConnections; 558 } 559 560 /** 561 * @param maximumConnections 562 * the maximumConnections to set 563 */ 564 public void setMaximumConnections(int maximumConnections) { 565 this.maximumConnections = maximumConnections; 566 } 567 568 @Override 569 public void started(Service service) { 570 } 571 572 @Override 573 public void stopped(Service service) { 574 this.currentTransportCount.decrementAndGet(); 575 } 576 577 @Override 578 public boolean isSslServer() { 579 return false; 580 } 581 582 @Override 583 public boolean isAllowLinkStealing() { 584 return allowLinkStealing; 585 } 586 587 @Override 588 public void setAllowLinkStealing(boolean allowLinkStealing) { 589 this.allowLinkStealing = allowLinkStealing; 590 } 591}