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}