import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
@GET @Path("/{itemname: [a-zA-Z_0-9]*}/state") 
@Produces( { MediaType.TEXT_PLAIN })
   public SuspendResponse<String> getPlainItemState(
   		@PathParam("itemname") String itemname, 
   		@Context AtmosphereResource resource) {
   	if(TRANSPORT.UNDEFINED.equals(resource.transport())) {
    	Item item = getItem(itemname);
    	if(item!=null) {
			if (logger.isDebugEnabled()) logger.debug("Received HTTP GET request at '{}'.", uriInfo.getPath());
			throw new WebApplicationException(Response.ok(item.getState().toString()).build());
    	} else {
    		if (logger.isDebugEnabled()) logger.info("Received HTTP GET request at '{}' for the unknown item '{}'.", uriInfo.getPath(), itemname);
    		throw new WebApplicationException(404);
    	}
	}
   	
   	BroadcasterFactory broadcasterFactory = resource.getAtmosphereConfig().getBroadcasterFactory();
   	GeneralBroadcaster itemBroadcaster = (GeneralBroadcaster) broadcasterFactory.lookup(GeneralBroadcaster.class, resource.getRequest().getPathInfo(), true); 
	itemBroadcaster.addStateChangeListener(new ItemStateChangeListener());
	return new SuspendResponse.SuspendResponseBuilder<String>()
			.scope(SCOPE.REQUEST)
			.resumeOnBroadcast(!ResponseTypeHelper.isStreamingTransport(resource.getRequest()))
			.broadcaster(itemBroadcaster)
			.outputComments(true).build();
   }
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
/**
 * Gets a list of all channel information.
 *
 * @return list of all channel informations
 */
public List<ChannelInformation> listChannels() {
    LinkedList<ChannelInformation> entries = new LinkedList<ChannelInformation>();
    Collection<Broadcaster> broadcasters = BroadcasterFactory.getDefault().lookupAll();
    String name;
    for (Broadcaster broadcaster : broadcasters) {
        if (broadcaster instanceof BounceProxyBroadcaster) {
            name = ((BounceProxyBroadcaster) broadcaster).getName();
        } else {
            name = broadcaster.getClass().getSimpleName();
        }

        Integer cachedSize = null;
        entries.add(new ChannelInformation(name, broadcaster.getAtmosphereResources().size(), cachedSize));
    }

    return entries;
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
public Broadcastable openChannel(String ccid, String atmosphereTrackingId) {

        throwExceptionIfTrackingIdnotSet(atmosphereTrackingId);

        log.debug("GET Channels open long poll channelId: {} trackingId: {}", ccid, atmosphereTrackingId);
        // NOTE: as of Atmosphere 0.8.5: even though the parameter is set
        // not to create the broadcaster if not
        // found, if the
        // broadcaster is found, but set to "destroyed" then it is recreated
        // TODO when is a broadcaster "destroyed" ???
        Broadcaster broadcaster = BroadcasterFactory.getDefault().lookup(BounceProxyBroadcaster.class, ccid, false);
        if (broadcaster == null) {
            log.error("no broadcaster registered for channel {}", ccid);
            // broadcaster not found for given ccid
            throw new JoynrHttpException(Status.BAD_REQUEST, JOYNRMESSAGINGERROR_CHANNELNOTFOUND);
        }

        // this causes the long poll, or immediate response if elements are
        // in the cache
        return new Broadcastable(broadcaster);
    }
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
@POST
@Override
public void broadcast(EventTemplate message) {
    if (api != null) {
        for (EnvironmentLogic e : api.environments().findAll()) {
            ZoneLogic z = e.getZone(message.getPayload().getStatementValue("zone.name"));
            if (z != null) {
                try {
                    BroadcasterFactory.getDefault()
                            .lookup("/" + RestAPIv3.API_VERSION + "/ws/" + AtmosphereZoneChangeResource.PATH)
                            .broadcast(
                                    om.writeValueAsString(z));
                } catch (JsonProcessingException ex) {

                }
                return;
            }
        }
    }
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
@Override
public void broadcast(EventTemplate message) {
    if (api != null) {
        String msg;
        try {
            EnvObjectLogic t = api.things().findOne(message.getPayload().getStatementValue("object.uuid"));
            if (t == null) {
                msg = "{}";
            } else {
                msg = om.writeValueAsString(t.getPojo());
            }
            BroadcasterFactory
                    .getDefault()
                    .lookup("/" + RestAPIv3.API_VERSION + "/ws/" + AtmosphereObjectChangeResource.PATH)
                    .broadcast(msg);
        } catch (JsonProcessingException ex) {
            LOG.warn(ex.getLocalizedMessage());
        }
    }
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
@POST
@Override
public void broadcast(EventTemplate message) {
    if (api != null) {
        for (Client c : api.getClients("plugin")) {
            Plugin p = (Plugin) c;
            if (p.getName().equalsIgnoreCase(message.getPayload().getStatementValue("plugin"))) {
                try {
                    BroadcasterFactory.getDefault()
                            .lookup("/" + RestAPIv3.API_VERSION + "/ws/" + AtmospherePluginChangeResource.PATH)
                            .broadcast(om.writeValueAsString(p));
                } catch (JsonProcessingException ex) {
                }
                return;
            }
        }
    }
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
public boolean sendNotification(String topicName, String message) throws IOException {
    BroadcasterFactory broadcasterFactory = framework.getBroadcasterFactory();
    Broadcaster b =  broadcasterFactory.lookup(Notifications.BASE_PATH + topicName);
    if(b != null) {
        b.broadcast(message);
        return true;
    } else {
        return false;
    }
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
private void broadcast(EventNotification eventNotification) {
    try {
        ServletContext servletContext = ServletContextFactory.getDefault().getServletContext();

        ((BroadcasterFactory) servletContext.getAttribute(BroadcasterFactory.class.getName())).lookup(broadcasterUUID)
                                                                                              .broadcast(mapper.writeValueAsString(eventNotification));
    } catch (Exception e) {
        log.error("Cannot broadcast event notification.", e);
        Throwables.propagate(e);
    }
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
@GET
@Produces({ MediaType.WILDCARD })
public SuspendResponse<Response> getResults(
		@Context HttpHeaders headers,
		@QueryParam("a") List<String> itemNames,
		@QueryParam("i") long index,
		@QueryParam("t") long time,
		@QueryParam("jsoncallback") @DefaultValue("callback") String callback,
   		@Context AtmosphereResource resource) {
	final String responseType = MediaTypeHelper.getResponseMediaType(headers.getAcceptableMediaTypes());
	if (logger.isDebugEnabled()) {
		logger.debug("Received HTTP GET request at '{}' for {} items at index '{}', ResponseType: '{}'.",
			uriInfo.getPath(), itemNames.size(), index, responseType);
	}
	List<ReturnType> rts = getReturnTypes(itemNames);
	if(index==0) {
		// first request => return all values
		if (responseType != null) {
			throw new WebApplicationException(Response.ok(getItemStateListBean(rts,System.currentTimeMillis()), responseType).build());
		} else {
			throw new WebApplicationException(Response.notAcceptable(null).build());
		}
	}
	
	BroadcasterFactory broadcasterFactory = resource.getAtmosphereConfig().getBroadcasterFactory();
	CometVisuBroadcaster itemBroadcaster = (CometVisuBroadcaster) broadcasterFactory.lookup(CometVisuBroadcaster.class, resource.getRequest().getPathInfo(), true);
	itemBroadcaster.addStateChangeListener(new ItemStateChangeListener(rts));
	
	return new SuspendResponse.SuspendResponseBuilder<Response>()
		.scope(SCOPE.REQUEST)
		.resumeOnBroadcast(!ResponseTypeHelper.isStreamingTransport(resource.getRequest()))
		.broadcaster(itemBroadcaster)
		.outputComments(true).build();
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
@GET @Path("/{itemname: [a-zA-Z_0-9]*}")
  @Produces( { MediaType.WILDCARD })
  public SuspendResponse<Response>  getItemData(
  		@Context HttpHeaders headers,
  		@PathParam("itemname") String itemname, 
  		@QueryParam("type") String type, 
  		@QueryParam("jsoncallback") @DefaultValue("callback") String callback,
  		@Context AtmosphereResource resource) {
if (logger.isDebugEnabled()) 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 ItemBean content = getItemDataBean(itemname);
    	final Object responseObject = ResponseHelper.wrapContentIfNeccessary(callback, responseType, content); 
    	throw new WebApplicationException(Response.ok(responseObject, responseType).build());  
	} else {
		throw new WebApplicationException(Response.notAcceptable(null).build());
	}
}

BroadcasterFactory broadcasterFactory = resource.getAtmosphereConfig().getBroadcasterFactory();
GeneralBroadcaster itemBroadcaster = (GeneralBroadcaster) broadcasterFactory.lookup(GeneralBroadcaster.class, resource.getRequest().getPathInfo(), true); 
itemBroadcaster.addStateChangeListener(new ItemStateChangeListener());

return new SuspendResponse.SuspendResponseBuilder<Response>()
			.scope(SCOPE.REQUEST)
			.resumeOnBroadcast(!ResponseTypeHelper.isStreamingTransport(resource.getRequest()))
			.broadcaster(itemBroadcaster)
			.outputComments(true).build(); 
  }
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
private void registerAtmosphere(Environment environment, Clock clock) {
    AtmosphereServlet servlet = new AtmosphereServlet();
    servlet.framework().addInitParameter(ApplicationConfig.ANNOTATION_PACKAGE, WebSocketHandler.class.getPackage().getName());
    servlet.framework().addInitParameter(ApplicationConfig.WEBSOCKET_SUPPORT, "true");

    ServletRegistration.Dynamic registration = environment.servlets().addServlet("atmosphere", servlet);
    registration.addMapping(WEBSOCKET_SERVLET_MAPPING);

    BroadcasterFactory broadcasterFactory = servlet.framework().getBroadcasterFactory();
    clock.setNextRoundCallback(round -> WebSocketHandler.broadcastRound(broadcasterFactory, round));
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
/**
 * Creates a long polling channel.
 *
 * @param ccid
 *            the identifier of the channel
 * @param atmosphereTrackingId
 *            the tracking ID of the channel
 * @return the path segment for the channel. The path, appended to the base
 *         URI of the channel service, can be used to post messages to the
 *         channel.
 */
public String createChannel(String ccid, String atmosphereTrackingId) {

    throwExceptionIfTrackingIdnotSet(atmosphereTrackingId);

    log.info("CREATE channel for cluster controller: {} trackingId: {} ", ccid, atmosphereTrackingId);
    Broadcaster broadcaster = null;
    // look for an existing broadcaster

    BroadcasterFactory defaultBroadcasterFactory = BroadcasterFactory.getDefault();
    if (defaultBroadcasterFactory == null) {
        throw new JoynrHttpException(500, 10009, "broadcaster was null");
    }

    broadcaster = defaultBroadcasterFactory.lookup(Broadcaster.class, ccid, false);
    // create a new one if none already exists
    if (broadcaster == null) {
        broadcaster = defaultBroadcasterFactory.get(BounceProxyBroadcaster.class, ccid);

    }

    // avoids error where previous long poll from browser got message
    // destined for new long poll
    // especially as seen in js, where every second refresh caused a fail
    for (AtmosphereResource resource : broadcaster.getAtmosphereResources()) {
        if (resource.uuid() != null && resource.uuid().equals(atmosphereTrackingId)) {
            resource.resume();
        }
    }

    UUIDBroadcasterCache broadcasterCache = (UUIDBroadcasterCache) broadcaster.getBroadcasterConfig()
                                                                              .getBroadcasterCache();
    broadcasterCache.activeClients().put(atmosphereTrackingId, System.currentTimeMillis());

    // BroadcasterCacheInspector is not implemented corrected in Atmosphere
    // 1.1.0RC4
    // broadcasterCache.inspector(new MessageExpirationInspector());

    return "/channels/" + ccid + "/";
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
/**
 * Deletes a channel from the broadcaster.
 *
 * @param ccid
 *            the channel to delete
 * @return <code>true</code> if the channel existed and could be deleted,
 *         <code>false</code> if there was no channel for the given ID and
 *         therefore could not be deleted.
 */
public boolean deleteChannel(String ccid) {
    log.info("DELETE channel for cluster controller: " + ccid);
    Broadcaster broadcaster = BroadcasterFactory.getDefault().lookup(Broadcaster.class, ccid, false);
    if (broadcaster == null) {
        return false;
    }

    BroadcasterFactory.getDefault().remove(ccid);

    broadcaster.resumeAll();
    broadcaster.destroy();
    // broadcaster.getBroadcasterConfig().forceDestroy();
    return true;
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
@Override
public void broadcast(EventTemplate message) {
    if (api != null) {
        String msg;
        try {
            msg = om.writeValueAsString(new MessageCalloutRepresentation(message.getProperty("message.text")));
            BroadcasterFactory
                    .getDefault()
                    .lookup("/" + RestAPIv3.API_VERSION + "/ws/" + AtmosphereMessageCalloutResource.PATH)
                    .broadcast(msg);
        } catch (JsonProcessingException ex) {
        }
    }
}
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的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(); 
   }
 

import org.atmosphere.cpr.BroadcasterFactory; //导入依赖的package包/类
/**
 * Posts a message to a long polling channel.
 *
 * @param ccid
 *            the identifier of the long polling channel
 * @param serializedMessage
 *            the message to send serialized as a SMRF message
 * @return the path segment for the message status. The path, appended to
 *         the base URI of the messaging service, can be used to query the
 *         message status
 *
 * @throws JoynrHttpException
 *             if one of:
 *             <ul>
 *             <li>ccid is not set</li>
 *             <li>the message has expired or not expiry date is set</li>
 *             <li>no channel registered for ccid</li>
 *             </ul>
 */
public String postMessage(String ccid, byte[] serializedMessage) {
    ImmutableMessage message;

    try {
        message = new ImmutableMessage(serializedMessage);
    } catch (EncodingException | UnsuppportedVersionException e) {
        throw new JoynrHttpException(Status.BAD_REQUEST, JOYNRMESSAGINGERROR_DESERIALIZATIONFAILED);
    }

    if (ccid == null) {
        log.error("POST message {} to cluster controller: NULL. Dropped because: channel Id was not set.",
                  message.getId());

        throw new JoynrHttpException(Status.BAD_REQUEST, JOYNRMESSAGINGERROR_CHANNELNOTSET);
    }

    // send the message to the receiver.
    if (message.getTtlMs() == 0) {
        log.error("POST message {} to cluster controller: {} dropped because: expiry date not set",
                  ccid,
                  message.getId());
        throw new JoynrHttpException(Status.BAD_REQUEST, JOYNRMESSAGINGERROR_EXPIRYDATENOTSET);
    }

    // Relative TTLs are not supported yet.
    if (!message.isTtlAbsolute()) {
        throw new JoynrHttpException(Status.BAD_REQUEST, JOYNRMESSAGINGERROR_RELATIVE_TTL_UNSPORTED);
    }

    if (message.getTtlMs() < System.currentTimeMillis()) {
        log.warn("POST message {} to cluster controller: {} dropped because: TTL expired", ccid, message.getId());
        throw new JoynrHttpException(Status.BAD_REQUEST, JOYNRMESSAGINGERROR_EXPIRYDATEEXPIRED);
    }

    // look for an existing broadcaster
    Broadcaster ccBroadcaster = BroadcasterFactory.getDefault().lookup(Broadcaster.class, ccid, false);
    if (ccBroadcaster == null) {
        // if the receiver has never registered with the bounceproxy
        // (or his registration has expired) then return 204 no
        // content.
        log.error("POST message {} to cluster controller: {} dropped because: no channel found",
                  ccid,
                  message.getId());
        throw new JoynrHttpException(Status.BAD_REQUEST, JOYNRMESSAGINGERROR_CHANNELNOTFOUND);
    }

    if (ccBroadcaster.getAtmosphereResources().size() == 0) {
        log.debug("no poll currently waiting for channelId: {}", ccid);
    }
    ccBroadcaster.broadcast(message);

    return "messages/" + message.getId();
}