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