import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
private HubMessage parseMessage(AtmosphereRequest request)
		throws IOException {
	JSONDeserializer<HubMessage> deserializer = new JSONDeserializer<HubMessage>();
	StringBuilder builder = new StringBuilder();
	BufferedReader reader = new BufferedReader(request.getReader());
	String line = reader.readLine();
	while (line != null) {
		builder.append(line);
		line = reader.readLine();
	}
	String stringMessage = builder.toString().trim();
	HubMessage msg = null;
	if (!"".equals(stringMessage)) {
		msg = deserializer.deserialize(stringMessage);
	}
	return msg;
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
/**
 * This is a middle level method to handle the details of Atmosphere and
 * http/ws level of processing request. It will eventually call
 * ApiFactory.process which handles the "pure" Jvm Java only processing
 * 
 * @param r
 *          r
 * @throws Exception
 *           e
 * 
 */
public void handleServiceApi(AtmosphereResource r) throws Exception {
  AtmosphereRequest request = r.getRequest();
  AtmosphereResponse response = r.getResponse();
  OutputStream out = response.getOutputStream();
  response.addHeader("Content-Type", CodecUtils.MIME_TYPE_JSON);
  
  String hack = null;
  byte[] data = null;
  if (request.body() != null) {
    hack = request.body().asString();
    // data = request.body().asBytes();
    if (hack != null){ // FIXME - hack because request.body().asBytes() ALWAYS returns null !!
      data = hack.getBytes();
    }
  }
  api.process(out, r.getRequest().getRequestURI(), data);
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@Override
public void onTextMessage(WebSocket webSocket, String data)
		throws IOException {
	AtmosphereResource r = webSocket.resource();
	AtmosphereRequest req = r.getRequest();
	String sessionId = req.getSession(true).getId();
	LOGGER.info("sessionId: {}, resourceId: {} : text message", sessionId,
			r.uuid());

	String json = data.trim();

	Player player = this.sessions.get(sessionId);
	LOGGER.info(player.toString());
	BaseClientGameCommand clientEvent = mapper.readValue(json,
			BaseClientGameCommand.class);
	clientEvent.setPlayer(player);
	clientEvent.setGameId(DEFAULT_GAME_ID);
	this.updater.issueCommand(DEFAULT_GAME_ID, clientEvent);

}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@Override
	public void onRequest(AtmosphereResource resource) throws IOException {
//		if (LOG.isTraceEnabled()) {
//			LOG.trace(String.format("onRequest() starts... for resource %s.", resource.uuid()));
//		}
		AtmosphereRequest request = resource.getRequest();

		String method = request.getMethod();
		if ("get".equalsIgnoreCase(method)) {
//			if (LOG.isTraceEnabled()) {
//				LOG.trace("Suspending request after GET call.");
//			}
			UserSession session = null;//sessionsPool.getSessionBySessionId(resource.uuid());
			if (session == null) {
				session = sessionsPool.createSession(resource);
				resource.suspend();
				resource.getResponse().setContentType("application/json");
				getMessageHandler().handleMessage(new MessageContext(sessionsPool, session, null, resource.getRequest().getLocale()));
			}
		} else if ("post".equalsIgnoreCase(method)) {
//			if (LOG.isTraceEnabled()) {
//				LOG.trace("Processing POST message.");
//			}
			HubMessage msg = parseMessage(request);
			if (msg == null) {
				return;
			}
			getMessageHandler().handleMessage(new MessageContext(
					sessionsPool, sessionsPool.getSessionBySessionId(resource.uuid()), msg, resource.getRequest().getLocale()));
		}
 	}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
public AtmosphereCoordinator route(AtmosphereRequest request, AtmosphereResponse response) throws IOException {
    final VertxAsyncIOWriter w = VertxAsyncIOWriter.class.cast(response.getAsyncIOWriter());
    try {

        Action a = framework.doCometSupport(request, response);
        final AtmosphereResourceImpl impl = (AtmosphereResourceImpl) request.getAttribute(FrameworkConfig.ATMOSPHERE_RESOURCE);

        String transport = (String) request.getAttribute(FrameworkConfig.TRANSPORT_IN_USE);
        if (transport == null) {
            transport = request.getHeader(X_ATMOSPHERE_TRANSPORT);
        }

        logger.debug("Transport {} action {}", transport, a);
        final Action action = (Action) request.getAttribute(NettyCometSupport.SUSPEND);
        if (action != null && action.type() == Action.TYPE.SUSPEND && action.timeout() != -1) {
            final AtomicReference<Future<?>> f = new AtomicReference();
            f.set(suspendTimer.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    if (!w.isClosed() && (System.currentTimeMillis() - w.lastTick()) > action.timeout()) {
                        asynchronousProcessor.endRequest(impl, false);
                        f.get().cancel(true);
                    }
                }
            }, action.timeout(), action.timeout(), TimeUnit.MILLISECONDS));
        }
    } catch (Throwable e) {
        logger.error("Unable to process request", e);
    }
    return this;
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@Override
public List<AtmosphereRequest> onTextStream(WebSocket webSocket, Reader data) {
    LOG.debug("processing reader message {}", data);
    String connectionKey = store.getConnectionKey(webSocket);
    consumer.sendMessage(connectionKey, data);
    LOG.debug("reader message sent");
    return null;
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@Override
public List<AtmosphereRequest> onBinaryStream(WebSocket webSocket, InputStream data) {
    LOG.debug("processing inputstream message {}", data);
    String connectionKey = store.getConnectionKey(webSocket);
    consumer.sendMessage(connectionKey, data);
    LOG.debug("reader message sent");
    return null;
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@Override
public List<AtmosphereRequest> onMessage(WebSocket webSocket, String data) {
    LOG.debug("processing text message {}", data);
    String connectionKey = store.getConnectionKey(webSocket);
    consumer.sendMessage(connectionKey, data);
    LOG.debug("text message sent");
    return null;
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
/**
 * handleMessagesApi handles all requests sent to /api/messages It
 * suspends/upgrades the connection to a websocket It is a asynchronous
 * protocol - and all messages are wrapped in a message wrapper.
 * 
 * The ApiFactory will handle the details of de-serializing but its necessary
 * to setup some protocol details here for websockets and session management
 * which the ApiFactory should not be concerned with.
 * 
 * @param r - request and response objects from Atmosphere server
 */
public void handleMessagesApi(AtmosphereResource r) {
  try {
    AtmosphereResponse response = r.getResponse();
    AtmosphereRequest request = r.getRequest();
    OutputStream out = response.getOutputStream();

    if (!r.isSuspended()) {
      r.suspend();
    }
    response.addHeader("Content-Type", CodecUtils.MIME_TYPE_JSON);

    api.process(this, out, r.getRequest().getRequestURI(), request.body().asString());

    /*
     * // FIXME - GET or POST should work - so this "should" be unnecessary ..
     * if ("GET".equals(httpMethod)) { // if post and parts.length < 3 ?? //
     * respond(r, apiKey, "getPlatform", new Hello(Runtime.getId()));
     * HashMap<URI, ServiceEnvironment> env = Runtime.getEnvironments();
     * respond(r, apiKey, "getLocalServices", env); } else {
     * doNotUseThisProcessMessageAPI(codec, request.body()); // FIXME data is
     * double encoded .. can't put '1st' encoded json // api.process(this,
     * out, r.getRequest().getRequestURI(), request.body().asString()); }
     */

  } catch (Exception e) {
    log.error("handleMessagesApi -", e);
  }
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
public void handleSession(AtmosphereResource r) {
  AtmosphereRequest request = r.getRequest();
  HttpSession s = request.getSession(true);
  if (s != null) {
    log.debug("put session uuid {}", r.uuid());
    sessions.put(r.uuid(), request.getSession(true));
  }
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@Override
public void onByteMessage(WebSocket webSocket, byte[] data, int offset,
		int length) throws IOException {
	AtmosphereResource r = webSocket.resource();
	AtmosphereRequest req = r.getRequest();
	String sessionId = req.getSession(true).getId();
	LOGGER.warn(
			"sessionId: {}, resourceId: {} : byte message should not happen",
			sessionId, r.uuid());
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@Override
public void onOpen(WebSocket webSocket) throws IOException {
	AtmosphereResource r = webSocket.resource();
	AtmosphereRequest req = r.getRequest();
	String sessionId = req.getSession(true).getId();
	LOGGER.info("sessionId: {}, resourceId: {} : websocket opened",
			sessionId, r.uuid());

	// debug, logging
	r.addEventListener(new WebSocketEventListenerImpl());
	// real listener
	r.addEventListener(new WebSocketDisconnectHandler());

	Player player = this.sessions.get(sessionId);

	// TODO: really it should just throw a runtime exception, it should not
	// happen at all, we need to check if a player exists, if not, then
	// redirect to the first page
	if (player == null) {
		LOGGER.info("New player");
		player = this.playerFactory.createRandomNamePlayer();
		this.sessions.put(sessionId, player);
	} else {
		this.connections.remove(player);
	}

	this.connections.put(player, r);

	NewPlayerCommand clientEvent = new NewPlayerCommand();
	clientEvent.setPlayer(player);
	clientEvent.setGameId(DEFAULT_GAME_ID);
	this.updater.issueCommand(DEFAULT_GAME_ID, clientEvent);
}
 

import org.atmosphere.cpr.AtmosphereRequest; //导入依赖的package包/类
@GET @Path("/{sitemapname: [a-zA-Z_0-9]*}/{pageid: [a-zA-Z_0-9]*}")
@Produces( { MediaType.WILDCARD })
   public SuspendResponse<Response> getPageData(
   		@Context HttpHeaders headers,
   		@PathParam("sitemapname") String sitemapname,
   		@PathParam("pageid") String pageId,
   		@QueryParam("type") String type, 
   		@QueryParam("jsoncallback") @DefaultValue("callback") String callback,
   		@Context AtmosphereResource resource) {
  		logger.debug("Received HTTP GET request at '{}' for media type '{}'.", uriInfo.getPath(), type);	
  		
	if(TRANSPORT.UNDEFINED.equals(resource.transport())) {
		final String responseType = MediaTypeHelper.getResponseMediaType(headers.getAcceptableMediaTypes(), type);
		if(responseType!=null) {
			final PageBean content = getPageBean(sitemapname, pageId, uriInfo.getBaseUriBuilder().build());
			final Object responseObject = ResponseHelper.wrapContentIfNeccessary(callback, responseType, content);
	    	throw new WebApplicationException(
	    			Response.ok(responseObject, responseType)
	    			.header(ATMOS_TIMEOUT_HEADER, DEFAULT_TIMEOUT_SECS + "")
	    			.build());
		} else {
			throw new WebApplicationException(Response.notAcceptable(null).build());
		}
	}
	
	BroadcasterFactory broadcasterFactory = resource.getAtmosphereConfig().getBroadcasterFactory();
	GeneralBroadcaster sitemapBroadcaster = broadcasterFactory.lookup(GeneralBroadcaster.class, resource.getRequest().getPathInfo(), true);
	sitemapBroadcaster.addStateChangeListener(new SitemapStateChangeListener());
	
	boolean resume = false;
	try {
		AtmosphereRequest request = resource.getRequest();
		resume = !ResponseTypeHelper.isStreamingTransport(request);
	} catch (Exception e) {
		logger.debug(e.getMessage(), e);
	}

	return new SuspendResponse.SuspendResponseBuilder<Response>()
		.scope(SCOPE.REQUEST)
		.resumeOnBroadcast(resume)
		.broadcaster(sitemapBroadcaster)
		.period(DEFAULT_TIMEOUT_SECS, TimeUnit.SECONDS)
		.outputComments(true).build(); 
   }