ZEPPELIN-331 Don't update back the browser where updated the angular object

zeppelin-server/src/main/java/org/apache/zeppelin/socket/NotebookServer.java

@@ -261,6 +261,26 @@ private void broadcast(String noteId, Message m) {
     }
   }
 
+  private void broadcastExcept(String noteId, Message m, NotebookSocket exclude) {
+    synchronized (noteSocketMap) {
+      List<NotebookSocket> socketLists = noteSocketMap.get(noteId);
+      if (socketLists == null || socketLists.size() == 0) {
+        return;
+      }
+      LOG.info("SEND >> " + m.op);
+      for (NotebookSocket conn : socketLists) {
+        if (exclude.equals(conn)) {
+          continue;
+        }
+        try {
+          conn.send(serializeMessage(m));
+        } catch (IOException e) {
+          LOG.error("socket error", e);
+        }
+      }
+    }
+  }
+
   private void broadcastAll(Message m) {
     synchronized (connectedSockets) {
       for (NotebookSocket conn : connectedSockets) {
@@ -425,7 +445,7 @@ private void updateParagraph(NotebookSocket conn, Notebook notebook,
     note.persist();
     broadcast(note.id(), new Message(OP.PARAGRAPH).put("paragraph", p));
   }
-  
+
   private void cloneNote(NotebookSocket conn, Notebook notebook, Message fromMessage)
       throws IOException, CloneNotSupportedException {
     String noteId = getOpenNoteId(conn);
@@ -475,7 +495,7 @@ private void completion(NotebookSocket conn, Notebook notebook,
    * @param notebook the notebook.
    * @param fromMessage the message.
    */
-  private void angularObjectUpdated(WebSocket conn, Notebook notebook,
+  private void angularObjectUpdated(NotebookSocket conn, Notebook notebook,
       Message fromMessage) {
     String noteId = (String) fromMessage.get("noteId");
     String interpreterGroupId = (String) fromMessage.get("interpreterGroupId");
@@ -529,20 +549,22 @@ private void angularObjectUpdated(WebSocket conn, Notebook notebook,
           if (interpreterGroupId.equals(setting.getInterpreterGroup().getId())) {
             AngularObjectRegistry angularObjectRegistry = setting
                 .getInterpreterGroup().getAngularObjectRegistry();
-            this.broadcast(
+            this.broadcastExcept(
                 n.id(),
                 new Message(OP.ANGULAR_OBJECT_UPDATE).put("angularObject", ao)
                     .put("interpreterGroupId", interpreterGroupId)
-                    .put("noteId", n.id()));
+                    .put("noteId", n.id()),
+                conn);
           }
         }
       }
     } else { // broadcast to all web session for the note
-      this.broadcast(
+      this.broadcastExcept(
           note.id(),
           new Message(OP.ANGULAR_OBJECT_UPDATE).put("angularObject", ao)
               .put("interpreterGroupId", interpreterGroupId)
-              .put("noteId", note.id()));
+              .put("noteId", note.id()),
+          conn);
     }
   }
 

zeppelin-server/src/test/java/org/apache/zeppelin/interpreter/mock/MockInterpreter1.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zeppelin.interpreter.mock;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+import org.apache.zeppelin.interpreter.Interpreter;
+import org.apache.zeppelin.interpreter.InterpreterContext;
+import org.apache.zeppelin.interpreter.InterpreterResult;
+import org.apache.zeppelin.scheduler.Scheduler;
+import org.apache.zeppelin.scheduler.SchedulerFactory;
+
+public class MockInterpreter1 extends Interpreter{
+  Map<String, Object> vars = new HashMap<String, Object>();
+
+  public MockInterpreter1(Properties property) {
+    super(property);
+  }
+
+  @Override
+  public void open() {
+  }
+
+  @Override
+  public void close() {
+  }
+
+  @Override
+  public InterpreterResult interpret(String st, InterpreterContext context) {
+    return new InterpreterResult(InterpreterResult.Code.SUCCESS, "repl1: "+st);
+  }
+
+  @Override
+  public void cancel(InterpreterContext context) {
+  }
+
+  @Override
+  public FormType getFormType() {
+    return FormType.SIMPLE;
+  }
+
+  @Override
+  public int getProgress(InterpreterContext context) {
+    return 0;
+  }
+
+  @Override
+  public Scheduler getScheduler() {
+    return SchedulerFactory.singleton().createOrGetFIFOScheduler("test_"+this.hashCode());
+  }
+
+  @Override
+  public List<String> completion(String buf, int cursor) {
+    return null;
+  }
+}

zeppelin-server/src/test/java/org/apache/zeppelin/socket/NotebookServerTest.java

@@ -20,16 +20,53 @@
 package org.apache.zeppelin.socket;
 
 import static org.junit.Assert.*;
+
 import java.io.IOException;
+
+import org.apache.zeppelin.interpreter.InterpreterGroup;
+import org.apache.zeppelin.interpreter.InterpreterSetting;
+import org.apache.zeppelin.notebook.Note;
+import org.apache.zeppelin.notebook.Notebook;
+import org.apache.zeppelin.rest.AbstractTestRestApi;
+import org.apache.zeppelin.server.ZeppelinServer;
+import org.apache.zeppelin.socket.Message.OP;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
 import org.junit.Test;
 
+import com.google.gson.Gson;
+
 import java.net.UnknownHostException;
 import java.net.InetAddress;
+import java.util.List;
+
+import javax.servlet.http.HttpServletRequest;
+
+import static org.mockito.Mockito.*;
+
 
 /**
  * BASIC Zeppelin rest api tests
  */
-public class NotebookServerTest {
+public class NotebookServerTest extends AbstractTestRestApi {
+
+
+  private static Notebook notebook;
+  private static NotebookServer notebookServer;
+  private static Gson gson;
+
+  @BeforeClass
+  public static void init() throws Exception {
+    AbstractTestRestApi.startUp();
+    gson = new Gson();
+    notebook = ZeppelinServer.notebook;
+    notebookServer = ZeppelinServer.notebookServer;
+  }
+
+  @AfterClass
+  public static void destroy() throws Exception {
+    AbstractTestRestApi.shutDown();
+  }
 
   @Test
   public void checkOrigin() throws UnknownHostException {
@@ -45,5 +82,68 @@ public void checkInvalidOrigin(){
     NotebookServer server = new NotebookServer();
     assertFalse(server.checkOrigin(new TestHttpServletRequest(), "http://evillocalhost:8080"));
   }
+
+  @Test
+  public void testMakeSureNoAngularObjectBroadcastToWebsocketWhoFireTheEvent() throws IOException {
+    // create a notebook
+    Note note1 = notebook.createNote();
+
+    // get reference to interpreterGroup
+    InterpreterGroup interpreterGroup = null;
+    List<InterpreterSetting> settings = note1.getNoteReplLoader().getInterpreterSettings();
+    for (InterpreterSetting setting : settings) {
+      if (setting.getInterpreterGroup() == null) {
+        continue;
+      }
+
+      interpreterGroup = setting.getInterpreterGroup();
+      break;
+    }
+
+    // add angularObject
+    interpreterGroup.getAngularObjectRegistry().add("object1", "value1", note1.getId());
+
+    // create two sockets and open it
+    NotebookSocket sock1 = createWebSocket();
+    NotebookSocket sock2 = createWebSocket();
+
+    assertEquals(sock1, sock1);
+    assertNotEquals(sock1, sock2);
+
+    notebookServer.onOpen(sock1);
+    notebookServer.onOpen(sock2);
+    verify(sock1, times(0)).send(anyString()); // getNote, getAngularObject
+    // open the same notebook from sockets
+    notebookServer.onMessage(sock1, gson.toJson(new Message(OP.GET_NOTE).put("id", note1.getId())));
+    notebookServer.onMessage(sock2, gson.toJson(new Message(OP.GET_NOTE).put("id", note1.getId())));
+
+    reset(sock1);
+    reset(sock2);
+
+    // update object from sock1
+    notebookServer.onMessage(sock1, gson.toJson(
+        new Message(OP.ANGULAR_OBJECT_UPDATED)
+        .put("noteId", note1.getId())
+        .put("name", "object1")
+        .put("value", "value1")
+        .put("interpreterGroupId", interpreterGroup.getId())));
+
+
+    // expect object is broadcasted except for where the update is created
+    verify(sock1, times(0)).send(anyString());
+    verify(sock2, times(1)).send(anyString());
+
+    notebook.removeNote(note1.getId());
+  }
+
+  private NotebookSocket createWebSocket() {
+    NotebookSocket sock = mock(NotebookSocket.class);
+    when(sock.getRequest()).thenReturn(createHttpServletRequest());
+    return sock;
+  }
+
+  private HttpServletRequest createHttpServletRequest() {
+    return mock(HttpServletRequest.class);
+  }
 }