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.juneau.rest.server.sse;
018
019import static org.apache.juneau.commons.utils.Shorts.*;
020import static org.apache.juneau.commons.utils.StringUtils.*;
021
022import java.util.concurrent.*;
023import java.util.logging.*;
024
025import org.apache.juneau.marshall.sse.*;
026
027/**
028 * In-memory server-side SSE broadcaster.
029 */
030@SuppressWarnings({
031   "resource" // SseSubscription returned by subscribe() is owned by the caller; the broadcaster only closes replaced or removed ones in subscribe()
032})
033public class SseBroadcaster {
034
035   private static final Logger LOGGER = Logger.getLogger(SseBroadcaster.class.getName());
036   private static final int DEFAULT_QUEUE_SIZE = 1024;
037
038   /**
039    * Creates a broadcaster with the default queue size.
040    *
041    * @return A new broadcaster.
042    */
043   public static SseBroadcaster create() {
044      return new SseBroadcaster(DEFAULT_QUEUE_SIZE);
045   }
046
047   private final int queueSize;
048   private final ConcurrentMap<String,SseSubscription> subscriptions;
049
050   /**
051    * Constructor.
052    *
053    * @param queueSize The per-subscriber queue size.
054    */
055   public SseBroadcaster(int queueSize) {
056      if (queueSize <= 0)
057         throw iaex("queueSize must be greater than 0.");
058      this.queueSize = queueSize;
059      subscriptions = new ConcurrentHashMap<>();
060   }
061
062   /**
063    * Creates or replaces a subscriber.
064    *
065    * @param id The subscriber identifier. Must not be <jk>null</jk> or empty.
066    * @return A new subscription.
067    */
068   public SseSubscription subscribe(String id) {
069      if (isEmpty(id))
070         throw iaex("id cannot be null or empty.");
071      var subscription = new SseSubscription(id, queueSize, this::removeSubscriber);
072      var previous = subscriptions.put(id, subscription);
073      if (previous != null)
074         previous.close();
075      return subscription;
076   }
077
078   /**
079    * Publishes an event to all active subscribers.
080    *
081    * @param event The event to publish. Can be <jk>null</jk> (ignored).
082    */
083   public void publish(SseEvent event) {
084      if (event == null)
085         return;
086      subscriptions.values().forEach(x -> {
087         if (x.offer(event))
088            LOGGER.fine(() -> "SSE queue overflow for subscriber " + x.getId() + ", dropped oldest event.");
089      });
090   }
091
092   void removeSubscriber(String id) {
093      // Called back from SseSubscription.close().  Only remove the entry when it is the closed subscription: when
094      // subscribe(id) replaces a subscriber, the map already holds the new one, which must stay open.
095      subscriptions.computeIfPresent(id, (k, v) -> v.isClosed() ? null : v);
096   }
097}