Merge remote-tracking branch 'origin/master'

Conflicts:
	apps/ifwd/src/main/java/org/onlab/onos/ifwd/IntentReactiveForwarding.java

 package org.onlab.onos.ifwd;
 
+import static org.slf4j.LoggerFactory.getLogger;
+
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -14,7 +16,6 @@ import org.onlab.onos.net.flow.TrafficSelector;
 import org.onlab.onos.net.flow.TrafficTreatment;
 import org.onlab.onos.net.host.HostService;
 import org.onlab.onos.net.intent.HostToHostIntent;
-import org.onlab.onos.net.intent.Intent;
 import org.onlab.onos.net.intent.IntentId;
 import org.onlab.onos.net.intent.IntentService;
 import org.onlab.onos.net.packet.DefaultOutboundPacket;
@@ -27,11 +28,6 @@ import org.onlab.onos.net.topology.TopologyService;
 import org.onlab.packet.Ethernet;
 import org.slf4j.Logger;
 
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-
-import static org.slf4j.LoggerFactory.getLogger;
-
 /**
  * WORK-IN-PROGRESS: Sample reactive forwarding application using intent framework.
  */
@@ -54,9 +50,7 @@ public class IntentReactiveForwarding {
 
     private ReactivePacketProcessor processor = new ReactivePacketProcessor();
 
-    private static long intentId = 1;
-
-    private Map<HostIdPair, IntentId> intents = new ConcurrentHashMap<>();
+    private static long intentId = 0x123000;
 
     @Activate
     public void activate() {
@@ -97,12 +91,8 @@ public class IntentReactiveForwarding {
                 return;
             }
 
-            // Install a new intent only if we have not installed one already
-            HostIdPair key = new HostIdPair(srcId, dstId);
-            if (!intents.containsKey(key)) {
-                // Otherwise forward and be done with it.
-                intents.put(key, setUpConnectivity(context, srcId, dstId).getId());
-            }
+            // Otherwise forward and be done with it.
+            setUpConnectivity(context, srcId, dstId);
             forwardPacketToDst(context, dst);
         }
     }
@@ -132,26 +122,15 @@ public class IntentReactiveForwarding {
     }
 
     // Install a rule forwarding the packet to the specified port.
-    private Intent setUpConnectivity(PacketContext context, HostId srcId, HostId dstId) {
+    private void setUpConnectivity(PacketContext context, HostId srcId, HostId dstId) {
         TrafficSelector selector = DefaultTrafficSelector.builder().build();
         TrafficTreatment treatment = DefaultTrafficTreatment.builder().build();
 
         HostToHostIntent intent =
                 new HostToHostIntent(new IntentId(intentId++), srcId, dstId,
                                      selector, treatment);
+
         intentService.submit(intent);
-        return intent;
     }
 
-
-    private class HostIdPair {
-        HostId one;
-        HostId two;
-
-        HostIdPair(HostId one, HostId two) {
-            boolean oneFirst = one.hashCode() < two.hashCode();
-            this.one = oneFirst ? one : two;
-            this.two = oneFirst ? two : one;
-        }
-    }
 }

@@ -2,4 +2,4 @@
  * Trivial application that provides simple form of reactive forwarding
  * using the intent service.
  */
-package org.onlab.onos.fwd;
+package org.onlab.onos.ifwd;

@@ -36,6 +36,12 @@
             <scope>test</scope>
         </dependency>
 
+	<dependency>
+	    <groupId>org.easymock</groupId>
+	    <artifactId>easymock</artifactId>
+	    <scope>test</scope>
+	</dependency>
+
         <!-- TODO Consider removing store dependency.
               Currently required for DistributedDeviceManagerTest. -->
         <dependency>

@@ -60,14 +60,15 @@ public class HostMonitor implements TimerTask {
      *
      * @param deviceService device service used to find edge ports
      * @param packetService packet service used to send packets on the data plane
-     * @param hostService host service used to look up host information
+     * @param hostManager host manager used to look up host information and
+     * probe existing hosts
      */
     public HostMonitor(DeviceService deviceService, PacketService packetService,
-            HostManager hostService) {
+            HostManager hostManager) {
 
         this.deviceService = deviceService;
         this.packetService = packetService;
-        this.hostManager = hostService;
+        this.hostManager = hostManager;
 
         monitoredAddresses = Collections.newSetFromMap(
                 new ConcurrentHashMap<IpAddress, Boolean>());

+package org.onlab.onos.net.host.impl;
+
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.verify;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+import org.junit.Test;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.Device;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Host;
+import org.onlab.onos.net.MastershipRole;
+import org.onlab.onos.net.Port;
+import org.onlab.onos.net.PortNumber;
+import org.onlab.onos.net.device.DeviceListener;
+import org.onlab.onos.net.device.DeviceService;
+import org.onlab.onos.net.flow.instructions.Instruction;
+import org.onlab.onos.net.flow.instructions.Instructions.OutputInstruction;
+import org.onlab.onos.net.host.HostProvider;
+import org.onlab.onos.net.host.PortAddresses;
+import org.onlab.onos.net.packet.OutboundPacket;
+import org.onlab.onos.net.packet.PacketProcessor;
+import org.onlab.onos.net.packet.PacketService;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.packet.ARP;
+import org.onlab.packet.Ethernet;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.IpPrefix;
+import org.onlab.packet.MacAddress;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Multimap;
+
+public class HostMonitorTest {
+
+    private IpAddress targetIpAddress = IpAddress.valueOf("10.0.0.1");
+    private IpPrefix targetIpPrefix = IpPrefix.valueOf(targetIpAddress.toOctets());
+
+    private IpPrefix sourcePrefix = IpPrefix.valueOf("10.0.0.99/24");
+    private MacAddress sourceMac = MacAddress.valueOf(1L);
+
+    private HostMonitor hostMonitor;
+
+    @Test
+    public void testMonitorHostExists() throws Exception {
+        ProviderId id = new ProviderId("fake://", "id");
+
+        Host host = createMock(Host.class);
+        expect(host.providerId()).andReturn(id);
+        replay(host);
+
+        HostManager hostManager = createMock(HostManager.class);
+        expect(hostManager.getHostsByIp(targetIpPrefix))
+                .andReturn(Collections.singleton(host));
+        replay(hostManager);
+
+        HostProvider hostProvider = createMock(HostProvider.class);
+        expect(hostProvider.id()).andReturn(id).anyTimes();
+        hostProvider.triggerProbe(host);
+        expectLastCall().once();
+        replay(hostProvider);
+
+        hostMonitor = new HostMonitor(null, null, hostManager);
+
+        hostMonitor.registerHostProvider(hostProvider);
+        hostMonitor.addMonitoringFor(targetIpAddress);
+
+        hostMonitor.run(null);
+
+        verify(hostProvider);
+    }
+
+    @Test
+    public void testMonitorHostDoesNotExist() throws Exception {
+        HostManager hostManager = createMock(HostManager.class);
+
+        DeviceId devId = DeviceId.deviceId("fake");
+
+        Device device = createMock(Device.class);
+        expect(device.id()).andReturn(devId).anyTimes();
+        replay(device);
+
+        PortNumber portNum = PortNumber.portNumber(1L);
+
+        Port port = createMock(Port.class);
+        expect(port.number()).andReturn(portNum).anyTimes();
+        replay(port);
+
+        TestDeviceService deviceService = new TestDeviceService();
+        deviceService.addDevice(device, Collections.singleton(port));
+
+        ConnectPoint cp = new ConnectPoint(devId, portNum);
+        PortAddresses pa = new PortAddresses(cp, Collections.singleton(sourcePrefix),
+                sourceMac);
+
+        expect(hostManager.getHostsByIp(targetIpPrefix))
+                .andReturn(Collections.<Host>emptySet()).anyTimes();
+        expect(hostManager.getAddressBindingsForPort(cp))
+                .andReturn(pa).anyTimes();
+        replay(hostManager);
+
+        TestPacketService packetService = new TestPacketService();
+
+
+        // Run the test
+        hostMonitor = new HostMonitor(deviceService, packetService, hostManager);
+
+        hostMonitor.addMonitoringFor(targetIpAddress);
+        hostMonitor.run(null);
+
+
+        // Check that a packet was sent to our PacketService and that it has
+        // the properties we expect
+        assertTrue(packetService.packets.size() == 1);
+        OutboundPacket packet = packetService.packets.get(0);
+
+        // Check the output port is correct
+        assertTrue(packet.treatment().instructions().size() == 1);
+        Instruction instruction = packet.treatment().instructions().get(0);
+        assertTrue(instruction instanceof OutputInstruction);
+        OutputInstruction oi = (OutputInstruction) instruction;
+        assertTrue(oi.port().equals(portNum));
+
+        // Check the output packet is correct (well the important bits anyway)
+        Ethernet eth = new Ethernet();
+        eth.deserialize(packet.data().array(), 0, packet.data().array().length);
+        ARP arp = (ARP) eth.getPayload();
+        assertTrue(Arrays.equals(arp.getSenderProtocolAddress(), sourcePrefix.toOctets()));
+        assertTrue(Arrays.equals(arp.getSenderHardwareAddress(), sourceMac.toBytes()));
+        assertTrue(Arrays.equals(arp.getTargetProtocolAddress(), targetIpPrefix.toOctets()));
+    }
+
+    class TestPacketService implements PacketService {
+
+        List<OutboundPacket> packets = new ArrayList<>();
+
+        @Override
+        public void addProcessor(PacketProcessor processor, int priority) {
+        }
+
+        @Override
+        public void removeProcessor(PacketProcessor processor) {
+        }
+
+        @Override
+        public void emit(OutboundPacket packet) {
+            packets.add(packet);
+        }
+    }
+
+    class TestDeviceService implements DeviceService {
+
+        List<Device> devices = Lists.newArrayList();
+        Multimap<DeviceId, Port> devicePorts = HashMultimap.create();
+
+        void addDevice(Device device, Set<Port> ports) {
+            devices.add(device);
+            for (Port p : ports) {
+                devicePorts.put(device.id(), p);
+            }
+        }
+
+        @Override
+        public int getDeviceCount() {
+            return 0;
+        }
+
+        @Override
+        public Iterable<Device> getDevices() {
+            return devices;
+        }
+
+        @Override
+        public Device getDevice(DeviceId deviceId) {
+            return null;
+        }
+
+        @Override
+        public MastershipRole getRole(DeviceId deviceId) {
+            return null;
+        }
+
+        @Override
+        public List<Port> getPorts(DeviceId deviceId) {
+            List<Port> ports = Lists.newArrayList();
+            for (Port p : devicePorts.get(deviceId)) {
+                ports.add(p);
+            }
+            return ports;
+        }
+
+        @Override
+        public Port getPort(DeviceId deviceId, PortNumber portNumber) {
+            return null;
+        }
+
+        @Override
+        public boolean isAvailable(DeviceId deviceId) {
+            return false;
+        }
+
+        @Override
+        public void addListener(DeviceListener listener) {
+        }
+
+        @Override
+        public void removeListener(DeviceListener listener) {
+        }
+    }
+}

@@ -3,6 +3,8 @@ package org.onlab.onos.store.cluster.messaging;
 import org.onlab.onos.cluster.ControllerNode;
 import org.onlab.onos.store.cluster.impl.ClusterNodesDelegate;
 
+// TODO: This service interface can be removed, once we properly start
+// using ClusterService
 /**
  * Service for administering communications manager.
  */

@@ -2,6 +2,8 @@ package org.onlab.onos.store.cluster.messaging;
 
 import org.onlab.onos.cluster.NodeId;
 
+// TODO: ClusterMessage should be aware about how to serialize the payload
+// TODO: Should payload type be made generic?
 /**
  * Base message for cluster-wide communications.
  */

@@ -10,4 +10,4 @@ public interface ClusterMessageHandler {
      * @param message cluster message.
      */
     public void handle(ClusterMessage message);
-}
\ No newline at end of file
+}

 package org.onlab.onos.store.cluster.messaging;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import java.util.Objects;
+
 /**
  * Representation of a message subject.
  * Cluster messages have associated subjects that dictate how they get handled
@@ -10,7 +14,7 @@ public class MessageSubject {
     private final String value;
 
     public MessageSubject(String value) {
-        this.value = value;
+        this.value = checkNotNull(value);
     }
 
     public String value() {
@@ -21,4 +25,24 @@ public class MessageSubject {
     public String toString() {
         return value;
     }
+
+    @Override
+    public int hashCode() {
+        return value.hashCode();
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj == null) {
+            return false;
+        }
+        if (getClass() != obj.getClass()) {
+            return false;
+        }
+        MessageSubject that = (MessageSubject) obj;
+        return Objects.equals(this.value, that.value);
+    }
 }

@@ -39,6 +39,7 @@ public class ClusterCommunicationManager
 
     private ControllerNode localNode;
     private ClusterNodesDelegate nodesDelegate;
+    // FIXME: `members` should go away and should be using ClusterService
     private Map<NodeId, ControllerNode> members = new HashMap<>();
     private final Timer timer = new Timer("onos-controller-heatbeats");
     public static final long HEART_BEAT_INTERVAL_MILLIS = 1000L;

@@ -3,6 +3,8 @@ package org.onlab.onos.store.cluster.messaging.impl;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 
 public final class ClusterMessageSubjects {
+    // avoid instantiation
     private ClusterMessageSubjects() {}
+
     public static final MessageSubject CLUSTER_MEMBERSHIP_EVENT = new MessageSubject("CLUSTER_MEMBERSHIP_EVENT");
 }

@@ -6,6 +6,8 @@ import java.util.Objects;
 
 import org.onlab.onos.store.Timestamp;
 
+import com.google.common.base.MoreObjects;
+
 /**
  * Wrapper class to store Timestamped value.
  * @param <T>
@@ -70,6 +72,14 @@ public final class Timestamped<T> {
         return Objects.equals(this.timestamp, that.timestamp);
     }
 
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                    .add("timestamp", timestamp)
+                    .add("value", value)
+                    .toString();
+    }
+
     // Default constructor for serialization
     @Deprecated
     protected Timestamped() {

@@ -122,6 +122,12 @@
                 <version>1.9.13</version>
             </dependency>
 
+	    <dependency>
+	        <groupId>org.easymock</groupId>
+	        <artifactId>easymock</artifactId>
+	        <version>3.2</version>
+	        <scope>test</scope>
+	    </dependency>
 
             <!-- Web related -->
             <dependency>

@@ -72,24 +72,23 @@ public final class MetricsManager implements MetricsService {
     private final CsvReporter reporter;
 
     public MetricsManager() {
-        this.componentsRegistry = new ConcurrentHashMap<>();
         this.metricsRegistry = new MetricRegistry();
-
         this.reporter = CsvReporter.forRegistry(metricsRegistry)
                 .formatFor(Locale.US)
                 .convertRatesTo(TimeUnit.SECONDS)
                 .convertDurationsTo(TimeUnit.MICROSECONDS)
-                .build(new File("/tmp/"));
-
-        reporter.start(10, TimeUnit.SECONDS);
+                .build(new File("/var/onos/log/metrics/"));
     }
 
     @Activate
     public void activate() {
+        this.componentsRegistry = new ConcurrentHashMap<>();
+        reporter.start(10, TimeUnit.SECONDS);
     }
 
     @Deactivate
     public void deactivate() {
+        reporter.stop();
     }
 
     /**

@@ -68,4 +68,4 @@ public class Endpoint {
         }
         return true;
     }
-}
\ No newline at end of file
+}

@@ -86,4 +86,4 @@ public final class InternalMessage implements Message {
             return message;
         }
     }
-}
\ No newline at end of file
+}

@@ -53,4 +53,4 @@ public class KryoSerializer implements Serializer {
     public void serialize(Object obj, ByteBuffer buffer) {
         serializerPool.serialize(obj, buffer);
     }
-}
\ No newline at end of file
+}

 /**
  * Asynchronous messaging APIs implemented using the Netty framework.
  */
-package org.onlab.netty;
\ No newline at end of file
+package org.onlab.netty;