Fixes for shared object concurrency

Author: mondain

pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>org.red5</groupId>
         <artifactId>red5-parent</artifactId>
-        <version>1.0.10-M10</version>
+        <version>1.0.10</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>red5-server-common</artifactId>
@@ -236,6 +236,13 @@
             <version>${tomcat.version}</version>
             <scope>provided</scope>
         </dependency>
+        <!-- future message bus work -->
+        <!-- 
+        <dependency>
+            <groupId>net.engio</groupId>
+            <artifactId>mbassador</artifactId>
+            <version>1.3.2</version>
+        </dependency> -->
         <dependency>
             <groupId>junit</groupId>
             <artifactId>junit</artifactId>

src/main/java/org/red5/server/AttributeStore.java

@@ -42,7 +42,7 @@ public class AttributeStore implements ICastingAttributeStore {
     /**
      * Map for attributes
      */
-    protected ConcurrentMap<String, Object> attributes = new ConcurrentAttributesMap<String, Object>(1);
+    protected ConcurrentMap<String, Object> attributes = new ConcurrentAttributesMap<>(1);
 
     /**
      * Creates empty attribute store. Object is not associated with a persistence storage.

src/main/java/org/red5/server/api/statistics/support/StatisticsCounter.java

@@ -21,28 +21,25 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
- * Counts numbers used by the statistics. Keeps track of current, maximum and total numbers.
+ * Counts numbers used by the statistics. Keeps track of current and total numbers.
  * 
  * @author The Red5 Project
  * @author Joachim Bauch (jojo@struktur.de)
  */
 public class StatisticsCounter {
 
-    /** Current number. */
+    /** Current number */
     private AtomicInteger current = new AtomicInteger();
 
-    /** Total number. */
-    private AtomicInteger total = new AtomicInteger();
-
-    /** Maximum number. */
-    private AtomicInteger max = new AtomicInteger();
+    /** Total number */
+    private int total;
 
     /**
      * Increment statistics by one.
      */
     public void increment() {
-        total.incrementAndGet();
-        max.compareAndSet(current.intValue(), current.incrementAndGet());
+        current.incrementAndGet();
+        total++;
     }
 
     /**
@@ -67,24 +64,25 @@ public int getCurrent() {
      * @return total
      */
     public int getTotal() {
-        return total.intValue();
+        return total;
     }
 
     /**
-     * Get maximum number.
+     * Get maximum number. Use total instead.
      * 
      * @return max
      */
+    @Deprecated
     public int getMax() {
-        return max.intValue();
+        return total;
     }
 
     /* (non-Javadoc)
      * @see java.lang.Object#toString()
      */
     @Override
     public String toString() {
-        return "StatisticsCounter [current=" + current + ", total=" + total + ", max=" + max + "]";
+        return "StatisticsCounter [current=" + current + ", total=" + total + "]";
     }
 
 }

src/main/java/org/red5/server/net/rtmp/RTMPConnection.java

@@ -27,7 +27,6 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.ScheduledFuture;
@@ -1571,7 +1570,7 @@ public long getPendingVideoMessages(Number streamId) {
      * @param events
      *            shared object events
      */
-    public void sendSharedObjectMessage(String name, int currentVersion, boolean persistent, ConcurrentLinkedQueue<ISharedObjectEvent> events) {
+    public void sendSharedObjectMessage(String name, int currentVersion, boolean persistent, Set<ISharedObjectEvent> events) {
         // create a new sync message for every client to avoid concurrent access through multiple threads
         SharedObjectMessage syncMessage = state.getEncoding() == Encoding.AMF3 ? new FlexSharedObjectMessage(null, name, currentVersion, persistent) : new SharedObjectMessage(null, name, currentVersion, persistent);
         syncMessage.addEvents(events);

src/main/java/org/red5/server/net/rtmp/event/BaseEvent.java

@@ -200,7 +200,7 @@ public void release() {
     protected abstract void releaseInternal();
 
     public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-        type = (Type) in.readObject();
+        type = Type.valueOf(in.readUTF());
         sourceType = in.readByte();
         timestamp = in.readInt();
         if (log.isTraceEnabled()) {
@@ -212,7 +212,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
         if (log.isTraceEnabled()) {
             log.trace("writeExternal - type: {} sourceType: {} timestamp: {}", type, sourceType, timestamp);
         }
-        out.writeObject(type);
+        out.writeUTF(type.name());
         out.writeByte(sourceType);
         out.writeInt(timestamp);
     }

src/main/java/org/red5/server/scope/Scope.java

@@ -622,17 +622,17 @@ public IScopeHandler getHandler() {
     /** {@inheritDoc} */
     @Deprecated
     public int getMaxClients() {
-        return connectionStats.getMax();
+        return connectionStats.getTotal();
     }
 
     /** {@inheritDoc} */
     public int getMaxConnections() {
-        return connectionStats.getMax();
+        return connectionStats.getTotal();
     }
 
     /** {@inheritDoc} */
     public int getMaxSubscopes() {
-        return subscopeStats.getMax();
+        return subscopeStats.getTotal();
     }
 
     /**

src/main/java/org/red5/server/so/ISharedObjectEvent.java

@@ -30,7 +30,7 @@ enum Type {
     /**
      * Returns the type of the event.
      * 
-     * @return the type of the event.
+     * @return the type of the event
      */
     public Type getType();
 
@@ -43,13 +43,7 @@ enum Type {
      * <li>the attribute name to delete for DELETE_ATTRIBUTE</li>
      * <li>the handler name to call for SEND_MESSAGE</li>
      * </ul>
-     * In all other cases the key is
-     * 
-     * <pre>
-     * null
-     * </pre>
-     * 
-     * .
+     * In all other cases the key is null
      * 
      * @return the key of the event
      */
@@ -63,13 +57,7 @@ enum Type {
      * <li>the attribute value to set for SET_ATTRIBUTE</li>
      * <li>a list of parameters to pass to the handler for SEND_MESSAGE</li>
      * </ul>
-     * In all other cases the value is
-     * 
-     * <pre>
-     * null
-     * </pre>
-     * 
-     * .
+     * In all other cases the value is null
      * 
      * @return the value of the event
      */

src/main/java/org/red5/server/so/ISharedObjectMessage.java

@@ -18,7 +18,7 @@
 
 package org.red5.server.so;
 
-import java.util.Queue;
+import java.util.Set;
 
 import org.red5.server.net.rtmp.event.IRTMPEvent;
 
@@ -53,28 +53,25 @@ public interface ISharedObjectMessage extends IRTMPEvent {
      * 
      * @return set of ISharedObjectEvents
      */
-    public Queue<ISharedObjectEvent> getEvents();
+    public Set<ISharedObjectEvent> getEvents();
 
     /**
      * Addition event handler
      * 
-     * @param type
-     *            Event type
-     * @param key
-     *            Handler key
-     * @param value
-     *            Event value (like arguments)
+     * @param type Event type
+     * @param key Handler key
+     * @param value Event value (like arguments)
      * @return true if event is added and false if it is not added
      */
     public boolean addEvent(ISharedObjectEvent.Type type, String key, Object value);
 
     /**
      * Add event handler
      * 
-     * @param event
-     *            SO event
+     * @param event SO event
+     * @return true if event is added and false if it is not added
      */
-    public void addEvent(ISharedObjectEvent event);
+    public boolean addEvent(ISharedObjectEvent event);
 
     /**
      * Clear shared object
@@ -84,17 +81,7 @@ public interface ISharedObjectMessage extends IRTMPEvent {
     /**
      * Is empty?
      *
-     * @return <pre>
-     * true
-     * </pre>
-     * 
-     *         if shared object is empty,
-     * 
-     *         <pre>
-     * false
-     * </pre>
-     * 
-     *         otherwise
+     * @return true if shared object is empty, false otherwise
      */
     public boolean isEmpty();