blob: d9064e86038b81a7d8eb115fb266ac7a7b185592 [file] [log] [blame]
/*
* 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 flex.messaging;
import flex.management.MBeanLifecycleManager;
import flex.management.MBeanServerLocatorFactory;
import flex.messaging.config.ConfigurationManager;
import flex.messaging.config.FlexConfigurationManager;
import flex.messaging.config.MessagingConfiguration;
import flex.messaging.endpoints.Endpoint;
import flex.messaging.io.SerializationContext;
import flex.messaging.io.TypeMarshallingContext;
import flex.messaging.log.HTTPRequestLog;
import flex.messaging.log.Log;
import flex.messaging.log.LogCategories;
import flex.messaging.log.Logger;
import flex.messaging.log.LoggingHttpServletRequestWrapper;
import flex.messaging.log.ServletLogTarget;
import flex.messaging.services.AuthenticationService;
import flex.messaging.util.ClassUtil;
import flex.messaging.util.ExceptionUtil;
import flex.messaging.util.Trace;
import javax.servlet.ServletConfig;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.security.Principal;
import java.util.concurrent.ConcurrentHashMap;
/**
* The MessageBrokerServlet bootstraps the MessageBroker,
* adds endpoints to it, and starts the broker. The servlet
* also acts as a facade for all http-based endpoints, in that
* the servlet receives the http request and then delegates to
* an endpoint that can handle the request's content type. This
* does not occur for non-http endpoints, such as the rtmp endpoint.
*
* @see flex.messaging.MessageBroker
*/
public class MessageBrokerServlet extends HttpServlet {
static final long serialVersionUID = -5293855229461612246L;
public static final String LOG_CATEGORY_STARTUP_BROKER = LogCategories.STARTUP_MESSAGEBROKER;
private static final String STRING_UNDEFINED_APPLICATION = "undefined";
private MessageBroker broker;
private HttpFlexSessionProvider httpFlexSessionProvider;
private static String FLEXDIR = "/WEB-INF/flex/";
private boolean log_errors = false;
/**
* Initializes the servlet in its web container, then creates
* the MessageBroker and adds Endpoints and Services to that broker.
* This servlet may keep a reference to an endpoint if it needs to
* delegate to it in the <code>service</code> method.
*/
public void init(ServletConfig servletConfig) throws ServletException {
super.init(servletConfig);
// allocate thread local variables
createThreadLocals();
// Set the servlet config as thread local
FlexContext.setThreadLocalObjects(null, null, null, null, null, servletConfig);
ServletLogTarget.setServletContext(servletConfig.getServletContext());
ClassLoader loader = getClassLoader();
if ("true".equals(servletConfig.getInitParameter("useContextClassLoader"))) {
loader = Thread.currentThread().getContextClassLoader();
}
// Should we wrap http request for later error logging?
log_errors = HTTPRequestLog.init(getServletContext());
// Start the broker
try {
// Get the configuration manager
ConfigurationManager configManager = loadMessagingConfiguration(servletConfig);
// Load configuration
MessagingConfiguration config = configManager.getMessagingConfiguration(servletConfig);
// Set up logging system ahead of everything else.
config.createLogAndTargets();
// Create broker.
broker = config.createBroker(servletConfig.getInitParameter("messageBrokerId"), loader);
// Set the servlet config as thread local
FlexContext.setThreadLocalObjects(null, null, broker, null, null, servletConfig);
setupPathResolvers();
// Set initial servlet context on broker
broker.setServletContext(servletConfig.getServletContext());
Logger logger = Log.getLogger(ConfigurationManager.LOG_CATEGORY);
if (Log.isInfo()) {
logger.info(VersionInfo.buildMessage());
}
// Create endpoints, services, security, and logger on the broker based on configuration
config.configureBroker(broker);
long timeBeforeStartup = 0;
if (Log.isDebug()) {
timeBeforeStartup = System.currentTimeMillis();
Log.getLogger(LOG_CATEGORY_STARTUP_BROKER).debug("MessageBroker with id '{0}' is starting.",
new Object[]{broker.getId()});
}
//initialize the httpSessionToFlexSessionMap
synchronized (HttpFlexSession.mapLock) {
if (servletConfig.getServletContext().getAttribute(HttpFlexSession.SESSION_MAP) == null)
servletConfig.getServletContext().setAttribute(HttpFlexSession.SESSION_MAP, new ConcurrentHashMap());
}
broker.start();
if (Log.isDebug()) {
long timeAfterStartup = System.currentTimeMillis();
Long diffMillis = timeAfterStartup - timeBeforeStartup;
Log.getLogger(LOG_CATEGORY_STARTUP_BROKER).debug("MessageBroker with id '{0}' is ready (startup time: '{1}' ms)",
new Object[]{broker.getId(), diffMillis});
}
// Report replaced tokens
configManager.reportTokens();
// Report any unused properties.
config.reportUnusedProperties();
// Setup provider for FlexSessions that wrap underlying J2EE HttpSessions.
httpFlexSessionProvider = new HttpFlexSessionProvider();
broker.getFlexSessionManager().registerFlexSessionProvider(HttpFlexSession.class, httpFlexSessionProvider);
// clear the broker and servlet config as this thread is done
FlexContext.clearThreadLocalObjects();
} catch (Throwable t) {
// On any unhandled exception destroy the broker, log it and rethrow.
String applicationName = servletConfig.getServletContext().getServletContextName();
if (applicationName == null)
applicationName = STRING_UNDEFINED_APPLICATION;
System.err.println("**** MessageBrokerServlet in application '" + applicationName
+ "' failed to initialize due to runtime exception: "
+ ExceptionUtil.exceptionFollowedByRootCausesToString(t));
destroy();
// We used to throw UnavailableException, but Weblogic didn't mark the webapp as failed. See bug FBR-237
throw new ServletException(t);
}
}
private void setupPathResolvers() {
setupExternalPathResolver();
setupInternalPathResolver();
}
private void setupExternalPathResolver() {
broker.setExternalPathResolver(
new MessageBroker.PathResolver() {
public InputStream resolve(String filename) throws FileNotFoundException {
return new FileInputStream(new File(filename));
}
}
);
}
private void setupInternalPathResolver() {
broker.setInternalPathResolver(
new MessageBroker.InternalPathResolver() {
public InputStream resolve(String filename) {
return getServletContext().getResourceAsStream(FLEXDIR + filename);
}
}
);
}
private static ConfigurationManager loadMessagingConfiguration(ServletConfig servletConfig) {
ConfigurationManager manager = null;
Class managerClass;
String className;
// Check for Custom Configuration Manager Specification
if (servletConfig != null) {
String p = servletConfig.getInitParameter("services.configuration.manager");
if (p != null) {
className = p.trim();
try {
managerClass = ClassUtil.createClass(className);
manager = (ConfigurationManager) managerClass.newInstance();
} catch (Throwable t) {
if (Trace.config) // Log is not initialized yet.
Trace.trace("Could not load configuration manager as: " + className);
}
}
}
if (manager == null) {
manager = new FlexConfigurationManager();
}
return manager;
}
/**
* Stops all endpoints in the MessageBroker, giving them a chance
* to perform any endpoint-specific clean up.
*/
public void destroy() {
if (broker != null) {
broker.stop();
if (broker.isManaged()) {
MBeanLifecycleManager.unregisterRuntimeMBeans(broker);
}
// release static thread locals
destroyThreadLocals();
}
}
/**
* Handle an incoming request, and delegate to an endpoint based on
* content type, if appropriate. The content type mappings for endpoints
* are not externally configurable, and currently the AmfEndpoint
* is the only delegate.
*/
public void service(HttpServletRequest req, HttpServletResponse res) {
if (log_errors) {
// Create a wrapper for the request object so we can save the body content
LoggingHttpServletRequestWrapper wrapper = new LoggingHttpServletRequestWrapper(req);
req = wrapper;
try {
// Read the body content
wrapper.doReadBody();
} catch (IOException ignore) {
// ignore, the wrapper will preserve what content we were able to read.
}
}
try {
// Update thread locals
broker.initThreadLocals();
// Set this first so it is in place for the session creation event. The
// current session is set by the FlexSession stuff right when it is available.
// The threadlocal FlexClient is set up during message deserialization in the
// MessageBrokerFilter.
FlexContext.setThreadLocalObjects(null, null, broker, req, res, getServletConfig());
HttpFlexSession fs = httpFlexSessionProvider.getOrCreateSession(req);
Principal principal;
if (FlexContext.isPerClientAuthentication()) {
principal = FlexContext.getUserPrincipal();
} else {
principal = fs.getUserPrincipal();
}
if (principal == null && req.getHeader("Authorization") != null) {
String encoded = req.getHeader("Authorization");
if (encoded.indexOf("Basic") > -1) {
encoded = encoded.substring(6); //Basic.length()+1
try {
((AuthenticationService) broker.getService(AuthenticationService.ID)).decodeAndLogin(encoded, broker.getLoginManager());
} catch (Exception e) {
if (Log.isDebug())
Log.getLogger(LogCategories.SECURITY).info("Authentication service could not decode and login: " + e.getMessage());
}
}
}
String contextPath = req.getContextPath();
String pathInfo = req.getPathInfo();
String endpointPath = req.getServletPath();
if (pathInfo != null)
endpointPath = endpointPath + pathInfo;
Endpoint endpoint;
try {
endpoint = broker.getEndpoint(endpointPath, contextPath);
} catch (MessageException me) {
if (Log.isInfo())
Log.getLogger(LogCategories.ENDPOINT_GENERAL).info("Received invalid request for endpoint path '{0}'.", new Object[]{endpointPath});
if (!res.isCommitted()) {
try {
res.sendError(HttpServletResponse.SC_NOT_FOUND);
} catch (IOException ignore) {
}
}
return;
}
try {
if (Log.isInfo()) {
Log.getLogger(LogCategories.ENDPOINT_GENERAL).info("Channel endpoint {0} received request.",
new Object[]{endpoint.getId()});
}
endpoint.service(req, res);
} catch (UnsupportedOperationException ue) {
if (Log.isInfo()) {
Log.getLogger(LogCategories.ENDPOINT_GENERAL).info("Channel endpoint {0} received request for an unsupported operation.",
new Object[]{endpoint.getId()},
ue);
}
if (!res.isCommitted()) {
try {
res.sendError(HttpServletResponse.SC_METHOD_NOT_ALLOWED);
} catch (IOException ignore) {
}
}
}
} catch (Throwable t) {
// Final resort catch block as recommended by Fortify as a potential System info leak
try {
Log.getLogger(LogCategories.ENDPOINT_GENERAL).error("Unexpected error encountered in Message Broker servlet", t);
res.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
} catch (IOException ignore) {
// ignore
}
} finally {
if (log_errors) {
String info = (String) req.getAttribute(HTTPRequestLog.HTTP_ERROR_INFO);
if (info != null) {
// Log the HttpRequest data
System.out.println("Exception occurred while processing HTTP request: " + info + ", request details logged in " + HTTPRequestLog.getFileName());
HTTPRequestLog.outputRequest(info, req);
}
}
FlexContext.clearThreadLocalObjects();
}
}
/**
* Hook for subclasses to override the class loader to use for loading user defined classes.
*
* @return the class loader for this class
*/
protected ClassLoader getClassLoader() {
return this.getClass().getClassLoader();
}
// Call ONLY on servlet startup
public static void createThreadLocals() {
// allocate static thread local objects
FlexContext.createThreadLocalObjects();
SerializationContext.createThreadLocalObjects();
TypeMarshallingContext.createThreadLocalObjects();
}
// Call ONLY on servlet shutdown
protected static void destroyThreadLocals() {
// clear static member variables
Log.clear();
MBeanServerLocatorFactory.clear();
// Destroy static thread local objects
FlexContext.releaseThreadLocalObjects();
SerializationContext.releaseThreadLocalObjects();
TypeMarshallingContext.releaseThreadLocalObjects();
}
}