[260908] Provide basic collaboration capabilities for all sessions of a repository

https://bugs.eclipse.org/bugs/show_bug.cgi?id=260908

8 files changed, 295 insertions, 13 deletions

diff --git a/plugins/org.eclipse.emf.cdo.server/.settings/.api_filters b/plugins/org.eclipse.emf.cdo.server/.settings/.api_filters
index a492541c0d..02392d2c2a 100644
--- a/plugins/org.eclipse.emf.cdo.server/.settings/.api_filters
+++ b/plugins/org.eclipse.emf.cdo.server/.settings/.api_filters
@@ -101,13 +101,6 @@
                 <message_argument value="Session(SessionManager, CDOServerProtocol, int, boolean)"/>
             </message_arguments>
         </filter>
-        <filter id="574660632">
-            <message_arguments>
-                <message_argument value="ISession"/>
-                <message_argument value="CDOProtocolSession"/>
-                <message_argument value="Session"/>
-            </message_arguments>
-        </filter>
     </resource>
     <resource path="src/org/eclipse/emf/cdo/internal/server/PackageManager.java" type="org.eclipse.emf.cdo.internal.server.PackageManager">
         <filter id="576720909">
@@ -133,11 +126,5 @@
                 <message_argument value="ISession"/>
             </message_arguments>
         </filter>
-        <filter id="574619656">
-            <message_arguments>
-                <message_argument value="CDOProtocolSession"/>
-                <message_argument value="ISession"/>
-            </message_arguments>
-        </filter>
     </resource>
 </component>
diff --git a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/Session.java b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/Session.java
index dd2d0dceda..115055cb5f 100644
--- a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/Session.java
+++ b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/Session.java
@@ -22,11 +22,13 @@ import org.eclipse.emf.cdo.common.id.CDOIDUtil;
 import org.eclipse.emf.cdo.common.model.CDOClass;
 import org.eclipse.emf.cdo.common.model.CDOFeature;
 import org.eclipse.emf.cdo.common.model.CDOPackageURICompressor;
+import org.eclipse.emf.cdo.common.protocol.CDOProtocolConstants;
 import org.eclipse.emf.cdo.common.revision.CDORevision;
 import org.eclipse.emf.cdo.common.revision.delta.CDORevisionDelta;
 import org.eclipse.emf.cdo.internal.server.bundle.OM;
 import org.eclipse.emf.cdo.internal.server.protocol.CDOServerProtocol;
 import org.eclipse.emf.cdo.internal.server.protocol.CommitNotificationRequest;
+import org.eclipse.emf.cdo.internal.server.protocol.RemoteSessionNotificationRequest;
 import org.eclipse.emf.cdo.server.IAudit;
 import org.eclipse.emf.cdo.server.ISession;
 import org.eclipse.emf.cdo.server.ITransaction;
@@ -84,6 +86,8 @@ public class Session extends Container<IView> implements ISession, CDOIDProvider
     }
   };
 
+  private boolean subscribed;
+
   /**
    * @since 2.0
    */
@@ -142,6 +146,29 @@ public class Session extends Container<IView> implements ISession, CDOIDProvider
   /**
    * @since 2.0
    */
+  public boolean isSubscribed()
+  {
+    return subscribed;
+  }
+
+  /**
+   * @since 2.0
+   */
+  public void setSubscribed(boolean subscribed)
+  {
+    checkActive();
+    if (this.subscribed != subscribed)
+    {
+      this.subscribed = subscribed;
+      byte opcode = subscribed ? CDOProtocolConstants.REMOTE_SESSION_SUBSCRIBED
+          : CDOProtocolConstants.REMOTE_SESSION_UNSUBSCRIBED;
+      sessionManager.handleRemoteSessionNotification(opcode, this);
+    }
+  }
+
+  /**
+   * @since 2.0
+   */
   public boolean isPassiveUpdateEnabled()
   {
     return passiveUpdateEnabled;
@@ -303,6 +330,29 @@ public class Session extends Container<IView> implements ISession, CDOIDProvider
     }
   }
 
+  /**
+   * @since 2.0
+   */
+  public void handleRemoteSessionNotification(byte opcode, ISession session)
+  {
+    try
+    {
+      IChannel channel = protocol.getChannel();
+      if (LifecycleUtil.isActive(channel))
+      {
+        new RemoteSessionNotificationRequest(channel, opcode, session).sendAsync();
+      }
+      else
+      {
+        OM.LOG.warn("Session channel is inactive: " + this);
+      }
+    }
+    catch (Exception ex)
+    {
+      OM.LOG.error(ex);
+    }
+  }
+
   public CDOID provideCDOID(Object idObject)
   {
     return (CDOID)idObject;
diff --git a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/SessionManager.java b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/SessionManager.java
index ebfa19f9f3..751185ad99 100644
--- a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/SessionManager.java
+++ b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/SessionManager.java
@@ -14,6 +14,7 @@ package org.eclipse.emf.cdo.internal.server;
 
 import org.eclipse.emf.cdo.common.id.CDOID;
 import org.eclipse.emf.cdo.common.id.CDOIDAndVersion;
+import org.eclipse.emf.cdo.common.protocol.CDOProtocolConstants;
 import org.eclipse.emf.cdo.common.revision.delta.CDORevisionDelta;
 import org.eclipse.emf.cdo.internal.server.bundle.OM;
 import org.eclipse.emf.cdo.internal.server.protocol.CDOServerProtocol;
@@ -118,6 +119,7 @@ public class SessionManager extends Container<ISession> implements ISessionManag
     }
 
     fireElementAddedEvent(session);
+    handleRemoteSessionNotification(CDOProtocolConstants.REMOTE_SESSION_OPENED, session);
     return session;
   }
 
@@ -133,6 +135,7 @@ public class SessionManager extends Container<ISession> implements ISessionManag
     if (removeSession != null)
     {
       fireElementRemovedEvent(session);
+      handleRemoteSessionNotification(CDOProtocolConstants.REMOTE_SESSION_CLOSED, session);
     }
   }
 
@@ -165,4 +168,18 @@ public class SessionManager extends Container<ISession> implements ISessionManag
       }
     }
   }
+
+  /**
+   * @since 2.0
+   */
+  public void handleRemoteSessionNotification(byte opcode, Session excludedSession)
+  {
+    for (Session session : getSessions())
+    {
+      if (session != excludedSession && session.isSubscribed())
+      {
+        session.handleRemoteSessionNotification(opcode, excludedSession);
+      }
+    }
+  }
 }
diff --git a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/CDOServerProtocol.java b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/CDOServerProtocol.java
index ed6b7afe96..70db8e6c23 100644
--- a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/CDOServerProtocol.java
+++ b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/CDOServerProtocol.java
@@ -124,6 +124,12 @@ public class CDOServerProtocol extends CDOProtocolImpl
     case CDOProtocolConstants.SIGNAL_OBJECT_LOCKED:
       return new ObjectLockedIndication(this);
 
+    case CDOProtocolConstants.SIGNAL_GET_REMOTE_SESSIONS:
+      return new GetRemoteSessionsIndication(this);
+
+    case CDOProtocolConstants.SIGNAL_UNSUBSCRIBE_REMOTE_SESSIONS:
+      return new UnsubscribeRemoteSessionsIndication(this);
+
     default:
       return super.createSignalReactor(signalID);
     }
diff --git a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/GetRemoteSessionsIndication.java b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/GetRemoteSessionsIndication.java
new file mode 100644
index 0000000000..865f77f565
--- /dev/null
+++ b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/GetRemoteSessionsIndication.java
@@ -0,0 +1,66 @@
+/**
+ * Copyright (c) 2004 - 2009 Eike Stepper (Berlin, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * 
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ */
+package org.eclipse.emf.cdo.internal.server.protocol;
+
+import org.eclipse.emf.cdo.common.io.CDODataInput;
+import org.eclipse.emf.cdo.common.io.CDODataOutput;
+import org.eclipse.emf.cdo.common.protocol.CDOProtocolConstants;
+import org.eclipse.emf.cdo.internal.server.Session;
+import org.eclipse.emf.cdo.internal.server.bundle.OM;
+
+import org.eclipse.net4j.util.om.trace.ContextTracer;
+
+import java.io.IOException;
+
+/**
+ * @author Eike Stepper
+ */
+public class GetRemoteSessionsIndication extends CDOReadIndication
+{
+  private static final ContextTracer PROTOCOL_TRACER = new ContextTracer(OM.DEBUG_PROTOCOL,
+      GetRemoteSessionsIndication.class);
+
+  private boolean subscribe;
+
+  public GetRemoteSessionsIndication(CDOServerProtocol protocol)
+  {
+    super(protocol, CDOProtocolConstants.SIGNAL_GET_REMOTE_SESSIONS);
+  }
+
+  @Override
+  protected void indicating(CDODataInput in) throws IOException
+  {
+    subscribe = in.readBoolean();
+    if (PROTOCOL_TRACER.isEnabled())
+    {
+      PROTOCOL_TRACER.format("Read subscribe: {0}", subscribe);
+    }
+  }
+
+  @Override
+  protected void responding(CDODataOutput out) throws IOException
+  {
+    Session localSession = getSession();
+    Session[] sessions = getSessionManager().getSessions();
+    for (Session session : sessions)
+    {
+      if (session != localSession)
+      {
+        out.writeInt(session.getSessionID());
+        out.writeString(session.getUserID());
+        out.writeBoolean(session.isSubscribed());
+      }
+    }
+
+    out.writeInt(CDOProtocolConstants.NO_MORE_REMOTE_SESSIONS);
+    localSession.setSubscribed(subscribe);
+  }
+}
diff --git a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/RemoteSessionNotificationRequest.java b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/RemoteSessionNotificationRequest.java
new file mode 100644
index 0000000000..8784503478
--- /dev/null
+++ b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/RemoteSessionNotificationRequest.java
@@ -0,0 +1,49 @@
+/**
+ * Copyright (c) 2004 - 2009 Eike Stepper (Berlin, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * 
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ *    Simon McDuff - http://bugs.eclipse.org/201266
+ *    Simon McDuff - http://bugs.eclipse.org/233490    
+ */
+package org.eclipse.emf.cdo.internal.server.protocol;
+
+import org.eclipse.emf.cdo.common.io.CDODataOutput;
+import org.eclipse.emf.cdo.common.protocol.CDOProtocolConstants;
+import org.eclipse.emf.cdo.server.ISession;
+
+import org.eclipse.net4j.channel.IChannel;
+
+import java.io.IOException;
+
+/**
+ * @author Eike Stepper
+ */
+public class RemoteSessionNotificationRequest extends CDOServerRequest
+{
+  private byte opcode;
+
+  private ISession session;
+
+  public RemoteSessionNotificationRequest(IChannel channel, byte opcode, ISession session)
+  {
+    super(channel, CDOProtocolConstants.SIGNAL_REMOTE_SESSION_NOTIFICATION);
+    this.opcode = opcode;
+    this.session = session;
+  }
+
+  @Override
+  protected void requesting(CDODataOutput out) throws IOException
+  {
+    out.writeByte(opcode);
+    out.writeInt(session.getSessionID());
+    if (opcode == CDOProtocolConstants.REMOTE_SESSION_OPENED)
+    {
+      out.writeString(session.getUserID());
+    }
+  }
+}
diff --git a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/UnsubscribeRemoteSessionsIndication.java b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/UnsubscribeRemoteSessionsIndication.java
new file mode 100644
index 0000000000..1c2b1cb7f8
--- /dev/null
+++ b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/internal/server/protocol/UnsubscribeRemoteSessionsIndication.java
@@ -0,0 +1,102 @@
+/**
+ * Copyright (c) 2004 - 2009 Eike Stepper (Berlin, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ * 
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ */
+package org.eclipse.emf.cdo.internal.server.protocol;
+
+import org.eclipse.emf.cdo.common.id.CDOID;
+import org.eclipse.emf.cdo.common.io.CDODataInput;
+import org.eclipse.emf.cdo.common.io.CDODataOutput;
+import org.eclipse.emf.cdo.common.protocol.CDOProtocolConstants;
+import org.eclipse.emf.cdo.common.revision.CDORevision;
+import org.eclipse.emf.cdo.internal.server.bundle.OM;
+import org.eclipse.emf.cdo.server.IAudit;
+import org.eclipse.emf.cdo.server.IView;
+
+import org.eclipse.net4j.util.om.trace.ContextTracer;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @author Eike Stepper
+ */
+public class UnsubscribeRemoteSessionsIndication extends CDOReadIndication
+{
+  private static final ContextTracer PROTOCOL_TRACER = new ContextTracer(OM.DEBUG_PROTOCOL, UnsubscribeRemoteSessionsIndication.class);
+
+  private List<CDORevision> revisions;
+
+  public UnsubscribeRemoteSessionsIndication(CDOServerProtocol protocol)
+  {
+    super(protocol, CDOProtocolConstants.SIGNAL_SET_AUDIT);
+  }
+
+  @Override
+  protected void indicating(CDODataInput in) throws IOException
+  {
+    int viewID = in.readInt();
+    if (PROTOCOL_TRACER.isEnabled())
+    {
+      PROTOCOL_TRACER.format("Read viewID: {0}", viewID);
+    }
+
+    long timeStamp = in.readLong();
+    if (PROTOCOL_TRACER.isEnabled())
+    {
+      PROTOCOL_TRACER.format("Read timeStamp: {0,date} {0,time}", timeStamp);
+    }
+
+    int size = in.readInt();
+    if (PROTOCOL_TRACER.isEnabled())
+    {
+      PROTOCOL_TRACER.format("Reading {0} IDs", size);
+    }
+
+    List<CDOID> invalidObjects = new ArrayList<CDOID>(size);
+    for (int i = 0; i < size; i++)
+    {
+      CDOID id = in.readCDOID();
+      invalidObjects.add(id);
+      if (PROTOCOL_TRACER.isEnabled())
+      {
+        PROTOCOL_TRACER.format("Read ID: {0}", id);
+      }
+    }
+
+    IView view = getSession().getView(viewID);
+    if (view instanceof IAudit)
+    {
+      IAudit audit = (IAudit)view;
+      revisions = audit.setTimeStamp(timeStamp, invalidObjects);
+    }
+  }
+
+  @Override
+  protected void responding(CDODataOutput out) throws IOException
+  {
+    if (PROTOCOL_TRACER.isEnabled())
+    {
+      PROTOCOL_TRACER.format("Writing {0} existanceFlags", revisions.size());
+    }
+
+    out.writeInt(revisions.size());
+    for (CDORevision revision : revisions)
+    {
+      boolean existanceFlag = revision != null;
+      if (PROTOCOL_TRACER.isEnabled())
+      {
+        PROTOCOL_TRACER.format("Writing existanceFlag: {0}", existanceFlag);
+      }
+
+      out.writeBoolean(existanceFlag);
+    }
+  }
+}
diff --git a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/server/ISession.java b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/server/ISession.java
index f7a1b22c01..cc6ba3300e 100644
--- a/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/server/ISession.java
+++ b/plugins/org.eclipse.emf.cdo.server/src/org/eclipse/emf/cdo/server/ISession.java
@@ -26,6 +26,11 @@ public interface ISession extends CDOCommonSession, IContainer<IView>
   /**
    * @since 2.0
    */
+  public boolean isSubscribed();
+
+  /**
+   * @since 2.0
+   */
   public IView openView(int viewID);
 
   /**