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.http; 018 019import java.net.InetSocketAddress; 020import java.net.URI; 021import java.util.Map; 022 023import org.apache.activemq.command.BrokerInfo; 024import org.apache.activemq.transport.SocketConnectorFactory; 025import org.apache.activemq.transport.WebTransportServerSupport; 026import org.apache.activemq.transport.util.TextWireFormat; 027import org.apache.activemq.transport.xstream.XStreamWireFormat; 028import org.apache.activemq.util.ServiceStopper; 029import org.eclipse.jetty.server.Connector; 030import org.eclipse.jetty.server.Handler; 031import org.eclipse.jetty.server.Server; 032import org.eclipse.jetty.servlet.ServletContextHandler; 033import org.eclipse.jetty.servlet.ServletHolder; 034 035public class HttpTransportServer extends WebTransportServerSupport { 036 037 private TextWireFormat wireFormat; 038 private final HttpTransportFactory transportFactory; 039 040 public HttpTransportServer(URI uri, HttpTransportFactory factory) { 041 super(uri); 042 this.bindAddress = uri; 043 this.transportFactory = factory; 044 socketConnectorFactory = new SocketConnectorFactory(); 045 } 046 047 @Override 048 public void setBrokerInfo(BrokerInfo brokerInfo) { 049 } 050 051 // Properties 052 // ------------------------------------------------------------------------- 053 public TextWireFormat getWireFormat() { 054 if (wireFormat == null) { 055 wireFormat = createWireFormat(); 056 } 057 return wireFormat; 058 } 059 060 public void setWireFormat(TextWireFormat wireFormat) { 061 this.wireFormat = wireFormat; 062 } 063 064 // Implementation methods 065 // ------------------------------------------------------------------------- 066 protected TextWireFormat createWireFormat() { 067 return new XStreamWireFormat(); 068 } 069 070 protected void setConnector(Connector connector) { 071 this.connector = connector; 072 } 073 074 @Override 075 protected void doStart() throws Exception { 076 createServer(); 077 if (connector == null) { 078 connector = socketConnectorFactory.createConnector(server); 079 } 080 081 URI boundTo = bind(); 082 083 ServletContextHandler contextHandler = 084 new ServletContextHandler(server, "/", ServletContextHandler.NO_SECURITY); 085 086 ServletHolder holder = new ServletHolder(); 087 holder.setServlet(new HttpTunnelServlet()); 088 contextHandler.addServlet(holder, "/"); 089 090 contextHandler.setAttribute("acceptListener", getAcceptListener()); 091 contextHandler.setAttribute("wireFormat", getWireFormat()); 092 contextHandler.setAttribute("transportFactory", transportFactory); 093 contextHandler.setAttribute("transportOptions", transportOptions); 094 095 addGzipHandler(contextHandler); 096 097 server.start(); 098 099 // Update the Connect To URI with our actual location in case the configured port 100 // was set to zero so that we report the actual port we are listening on. 101 102 int port = boundTo.getPort(); 103 int p2 = getConnectorLocalPort(); 104 if (p2 != -1) { 105 port = p2; 106 } 107 108 setConnectURI(new URI(boundTo.getScheme(), 109 boundTo.getUserInfo(), 110 boundTo.getHost(), 111 port, 112 boundTo.getPath(), 113 boundTo.getQuery(), 114 boundTo.getFragment())); 115 } 116 117 private int getConnectorLocalPort() throws Exception { 118 return (Integer)connector.getClass().getMethod("getLocalPort").invoke(connector); 119 } 120 private void addGzipHandler(ServletContextHandler contextHandler) throws Exception { 121 Handler handler = null; 122 try { 123 handler = (Handler)Class.forName("org.eclipse.jetty.server.handler.GzipHandler", true, Handler.class.getClassLoader()).newInstance(); 124 } catch (Throwable t) { 125 handler = (Handler)Class.forName("org.eclipse.jetty.servlets.gzip.GzipHandler", true, Handler.class.getClassLoader()).newInstance(); 126 } 127 contextHandler.setHandler(handler); 128 } 129 130 @Override 131 protected void doStop(ServiceStopper stopper) throws Exception { 132 Server temp = server; 133 server = null; 134 if (temp != null) { 135 temp.stop(); 136 } 137 } 138 139 @Override 140 public InetSocketAddress getSocketAddress() { 141 return null; 142 } 143 144 @Override 145 public void setTransportOption(Map<String, Object> transportOptions) { 146 socketConnectorFactory.setTransportOptions(transportOptions); 147 super.setTransportOption(transportOptions); 148 } 149 150 @Override 151 public boolean isSslServer() { 152 return false; 153 } 154}