import org.atmosphere.cpr.Broadcaster; //导入依赖的package包/类
/**
 * Initialize WebSocket based communication channel between the client and
 * the server.
 */
@GET
@Path("/events")
public String subscribe(@Context HttpServletRequest req, @HeaderParam("sessionid") String sessionId)
        throws NotConnectedRestException {
    checkAccess(sessionId);
    HttpSession session = checkNotNull(req.getSession(),
                                       "HTTP session object is null. HTTP session support is requried for REST Scheduler eventing.");
    AtmosphereResource atmosphereResource = checkNotNull((AtmosphereResource) req.getAttribute(AtmosphereResource.class.getName()),
                                                         "No AtmosphereResource is attached with current request.");
    // use session id as the 'topic' (or 'id') of the broadcaster
    session.setAttribute(ATM_BROADCASTER_ID, sessionId);
    session.setAttribute(ATM_RESOURCE_ID, atmosphereResource.uuid());
    Broadcaster broadcaster = lookupBroadcaster(sessionId, true);
    if (broadcaster != null) {
        atmosphereResource.setBroadcaster(broadcaster).suspend();
    }
    return null;
}
 

import org.atmosphere.cpr.Broadcaster; //导入依赖的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.Broadcaster; //导入依赖的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.Broadcaster; //导入依赖的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.Broadcaster; //导入依赖的package包/类
@Override
public Broadcaster initialize(String name, URI uri, AtmosphereConfig config) {
	super.initialize(name, uri, config);
	this.addBroadcasterLifeCyclePolicyListener(new BroadcasterLifeCyclePolicyListener() {
		
		@Override
		public void onIdle() {
			logger.debug("broadcaster '{}' is idle", this.toString());
		}
		
		@Override
		public void onEmpty() {
			logger.debug("broadcaster '{}' is empty", this.toString());
		}
		
		@Override
		public void onDestroy() {
			logger.debug("broadcaster '{}' destroyed", this.toString());
			for (ResourceStateChangeListener l : listeners){
				l.unregisterItems();
				listeners.remove(l);
			}
		}
	});
	CVApplication.modelRepository.addModelRepositoryChangeListener(this);
	return this;
}
 

import org.atmosphere.cpr.Broadcaster; //导入依赖的package包/类
@Override
public Broadcaster initialize(String name, URI uri, AtmosphereConfig config) {
	super.initialize(name, uri, config);
	
	this.addBroadcasterLifeCyclePolicyListener(new BroadcasterLifeCyclePolicyListener() {
		
		@Override
		public void onIdle() {
			logger.debug("broadcaster '{}' is idle", this.toString());
		}
		
		@Override
		public void onEmpty() {
			logger.debug("broadcaster '{}' is empty", this.toString());
		}
		
		@Override
		public void onDestroy() {
			logger.debug("broadcaster '{}' destroyed", this.toString());
			logger.trace("broadcaster '{}' left {} {} instaces", this.toString(), listeners.size(), ResourceStateChangeListener.class.getName());
			for (ResourceStateChangeListener listener : listeners){
				listener.unregisterItems();
				boolean removed = listeners.remove(listener);
				if (!removed) logger.warn("Could not remove event listener '{}', this may cause a memory leak.", listener.toString());
			}
		}
	});
	
	return this;
}
 

import org.atmosphere.cpr.Broadcaster; //导入依赖的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.Broadcaster; //导入依赖的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.Broadcaster; //导入依赖的package包/类
private Broadcaster lookupBroadcaster(String topic, boolean createNew) {
    AtmosphereResource atmosphereResource = (AtmosphereResource) httpServletRequest.getAttribute("org.atmosphere.cpr.AtmosphereResource");
    return atmosphereResource.getAtmosphereConfig().getBroadcasterFactory().lookup(topic, createNew);
}
 

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