Top-k monitoring implementation

topk-plugin/README.md

@@ -0,0 +1,2 @@
+Distributed top-k monitoring (Babcock, Olston SIGMOD 2003)
+on top of the Infispan Communication layer.

topk-plugin/pom.xml

@@ -0,0 +1,97 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>eu.leads</groupId>
+    <artifactId>distsum</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <name></name>
+    <organization>
+        <name>The LEADS project</name>
+        <url>http://www.leads-project.eu/</url>
+    </organization>
+
+    <properties>
+        <infinispan.version>7.0.0.Alpha4</infinispan.version>
+    </properties>
+
+    <dependencies>
+
+        <!-- Infinispan Dependencies -->
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-core</artifactId>
+            <version>${infinispan.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-client-hotrod</artifactId>
+            <version>${infinispan.version}</version>
+        </dependency>
+
+        <!-- test Dependencies -->
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <scope>test</scope>
+            <version>6.8</version>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-core</artifactId>
+            <version>${infinispan.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-client-hotrod</artifactId>
+            <version>${infinispan.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-server-core</artifactId>
+            <version>${infinispan.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-server-hotrod</artifactId>
+            <version>${infinispan.version}</version>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.infinispan</groupId>
+            <artifactId>infinispan-server-hotrod</artifactId>
+            <version>${infinispan.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>12.0</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.1</version>
+                <configuration>
+                    <source>1.6</source>
+                    <target>1.6</target>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

topk-plugin/src/main/java/eu/leads/distsum/ComChannel.java

@@ -0,0 +1,58 @@
+package eu.leads.distsum;
+
+import org.infinispan.Cache;
+import org.infinispan.filter.KeyValueFilter;
+import org.infinispan.metadata.Metadata;
+
+/**
+ *
+ * @author vagvaz
+ * @author otrack
+ *
+ * Created by vagvaz on 7/5/14.
+ *
+ * Communication channel just just a map of String,Node to send messages
+ */
+public class ComChannel {
+
+    private Cache<String,Message> nodes;
+
+    public ComChannel(Cache<String,Message> c) {
+        nodes =  c;
+    }
+
+    // Add a node to the map
+    public void register(final String id, Node node){
+        KeyValueFilter<String,Message> filter= new KeyValueFilter<String,Message>(){
+            public boolean accept(String i, Message msg, Metadata metadata){
+                return id.equals(i);
+            }
+        };
+        nodes.put(id, Message.EMPTYMSG); // to create the entry
+        nodes.addListener(node, filter, null);
+    }
+
+    //Send messsage to node id
+    public void sentTo(String id, Message message){
+        nodes.put(id, message);
+    }
+
+    // Broadcast a message to all nodes, but coordinator
+    // The coordinator takes as a result the replies of the nodes
+    /*public void broadCast(Message message){
+        for(String node: nodes.keySet()){
+            if(!node.equals(Node.COORDINATOR)){
+                nodes.put(node,message);
+            }
+        }
+    }*/
+
+    /*public void broadCastToAllButOne(Message message, String node1){
+        for(String node: nodes.keySet()){
+            if( (!node.equals(Node.COORDINATOR)) && (!node.equals(node1)) ){
+                nodes.put(node,message);
+            }
+        }
+    }*/
+
+}