added onos-start-network and fake distributed stores and fixed mastership bugs/omissions

Change-Id: I6cf9f1a13bf9e8f715bf5d72249431d94878b204

 package org.onlab.onos.cluster.impl;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.Set;
+
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -12,6 +17,7 @@ import org.onlab.onos.cluster.MastershipEvent;
 import org.onlab.onos.cluster.MastershipListener;
 import org.onlab.onos.cluster.MastershipService;
 import org.onlab.onos.cluster.MastershipStore;
+import org.onlab.onos.cluster.MastershipStoreDelegate;
 import org.onlab.onos.cluster.MastershipTerm;
 import org.onlab.onos.cluster.MastershipTermService;
 import org.onlab.onos.cluster.NodeId;
@@ -21,15 +27,10 @@ import org.onlab.onos.net.DeviceId;
 import org.onlab.onos.net.MastershipRole;
 import org.slf4j.Logger;
 
-import java.util.Set;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-import static org.slf4j.LoggerFactory.getLogger;
-
 @Component(immediate = true)
 @Service
 public class MastershipManager
-        implements MastershipService, MastershipAdminService {
+implements MastershipService, MastershipAdminService {
 
     private static final String NODE_ID_NULL = "Node ID cannot be null";
     private static final String DEVICE_ID_NULL = "Device ID cannot be null";
@@ -38,7 +39,9 @@ public class MastershipManager
     private final Logger log = getLogger(getClass());
 
     protected final AbstractListenerRegistry<MastershipEvent, MastershipListener>
-            listenerRegistry = new AbstractListenerRegistry<>();
+    listenerRegistry = new AbstractListenerRegistry<>();
+
+    private final MastershipStoreDelegate delegate = new InternalDelegate();
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected MastershipStore store;
@@ -52,12 +55,14 @@ public class MastershipManager
     @Activate
     public void activate() {
         eventDispatcher.addSink(MastershipEvent.class, listenerRegistry);
+        store.setDelegate(delegate);
         log.info("Started");
     }
 
     @Deactivate
     public void deactivate() {
         eventDispatcher.removeSink(MastershipEvent.class);
+        store.unsetDelegate(delegate);
         log.info("Stopped");
     }
 
@@ -141,4 +146,14 @@ public class MastershipManager
 
     }
 
+    public class InternalDelegate implements MastershipStoreDelegate {
+
+        @Override
+        public void notify(MastershipEvent event) {
+            log.info("dispatching mastership event {}", event);
+            eventDispatcher.post(event);
+        }
+
+    }
+
 }

 package org.onlab.onos.net.device.impl;
 
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.onlab.onos.net.device.DeviceEvent.Type.DEVICE_MASTERSHIP_CHANGED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.List;
+
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -32,20 +38,14 @@ import org.onlab.onos.net.provider.AbstractProviderRegistry;
 import org.onlab.onos.net.provider.AbstractProviderService;
 import org.slf4j.Logger;
 
-import java.util.List;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-import static org.onlab.onos.net.device.DeviceEvent.Type.DEVICE_MASTERSHIP_CHANGED;
-import static org.slf4j.LoggerFactory.getLogger;
-
 /**
  * Provides implementation of the device SB & NB APIs.
  */
 @Component(immediate = true)
 @Service
 public class DeviceManager
-        extends AbstractProviderRegistry<DeviceProvider, DeviceProviderService>
-        implements DeviceService, DeviceAdminService, DeviceProviderRegistry {
+extends AbstractProviderRegistry<DeviceProvider, DeviceProviderService>
+implements DeviceService, DeviceAdminService, DeviceProviderRegistry {
 
     private static final String DEVICE_ID_NULL = "Device ID cannot be null";
     private static final String PORT_NUMBER_NULL = "Port number cannot be null";
@@ -56,9 +56,9 @@ public class DeviceManager
     private final Logger log = getLogger(getClass());
 
     protected final AbstractListenerRegistry<DeviceEvent, DeviceListener>
-            listenerRegistry = new AbstractListenerRegistry<>();
+    listenerRegistry = new AbstractListenerRegistry<>();
 
-    private DeviceStoreDelegate delegate = new InternalStoreDelegate();
+    private final DeviceStoreDelegate delegate = new InternalStoreDelegate();
 
     private final MastershipListener mastershipListener = new InternalMastershipListener();
 
@@ -170,8 +170,8 @@ public class DeviceManager
 
     // Personalized device provider service issued to the supplied provider.
     private class InternalDeviceProviderService
-            extends AbstractProviderService<DeviceProvider>
-            implements DeviceProviderService {
+    extends AbstractProviderService<DeviceProvider>
+    implements DeviceProviderService {
 
         InternalDeviceProviderService(DeviceProvider provider) {
             super(provider);
@@ -183,14 +183,14 @@ public class DeviceManager
             checkNotNull(deviceDescription, DEVICE_DESCRIPTION_NULL);
             checkValidity();
             DeviceEvent event = store.createOrUpdateDevice(provider().id(),
-                                                           deviceId, deviceDescription);
+                    deviceId, deviceDescription);
 
             // If there was a change of any kind, trigger role selection process.
             if (event != null) {
                 log.info("Device {} connected", deviceId);
                 mastershipService.requestRoleFor(deviceId);
                 provider().roleChanged(event.subject(),
-                                       mastershipService.getLocalRole(deviceId));
+                        mastershipService.getLocalRole(deviceId));
                 post(event);
             }
         }
@@ -225,7 +225,7 @@ public class DeviceManager
             DeviceEvent event = store.updatePortStatus(deviceId, portDescription);
             if (event != null) {
                 log.info("Device {} port {} status changed", deviceId,
-                         event.port().number());
+                        event.port().number());
                 post(event);
             }
         }
@@ -249,9 +249,10 @@ public class DeviceManager
     private class InternalMastershipListener implements MastershipListener {
         @Override
         public void event(MastershipEvent event) {
-            // FIXME: for now we're taking action only on becoming master
             if (event.master().equals(clusterService.getLocalNode().id())) {
                 applyRole(event.subject(), MastershipRole.MASTER);
+            } else {
+                applyRole(event.subject(), MastershipRole.STANDBY);
             }
         }
     }

 package org.onlab.onos.store.cluster.impl;
 
-import com.google.common.base.Optional;
-import com.google.common.cache.LoadingCache;
-import com.google.common.collect.ImmutableSet;
-import com.hazelcast.core.IMap;
+import static com.google.common.cache.CacheBuilder.newBuilder;
+import static org.onlab.onos.cluster.MastershipEvent.Type.MASTER_CHANGED;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
@@ -23,11 +25,10 @@ import org.onlab.onos.store.impl.AbsentInvalidatingLoadingCache;
 import org.onlab.onos.store.impl.AbstractDistributedStore;
 import org.onlab.onos.store.impl.OptionalCacheLoader;
 
-import java.util.Map;
-import java.util.Objects;
-import java.util.Set;
-
-import static com.google.common.cache.CacheBuilder.newBuilder;
+import com.google.common.base.Optional;
+import com.google.common.cache.LoadingCache;
+import com.google.common.collect.ImmutableSet;
+import com.hazelcast.core.IMap;
 
 /**
  * Distributed implementation of the cluster nodes store.
@@ -35,8 +36,8 @@ import static com.google.common.cache.CacheBuilder.newBuilder;
 @Component(immediate = true)
 @Service
 public class DistributedMastershipStore
-        extends AbstractDistributedStore<MastershipEvent, MastershipStoreDelegate>
-        implements MastershipStore {
+extends AbstractDistributedStore<MastershipEvent, MastershipStoreDelegate>
+implements MastershipStore {
 
     private IMap<byte[], byte[]> rawMasters;
     private LoadingCache<DeviceId, Optional<NodeId>> masters;
@@ -51,9 +52,9 @@ public class DistributedMastershipStore
 
         rawMasters = theInstance.getMap("masters");
         OptionalCacheLoader<DeviceId, NodeId> nodeLoader
-                = new OptionalCacheLoader<>(storeService, rawMasters);
+        = new OptionalCacheLoader<>(storeService, rawMasters);
         masters = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
-        rawMasters.addEntryListener(new RemoteEventHandler<>(masters), true);
+        rawMasters.addEntryListener(new RemoteMasterShipEventHandler(masters), true);
 
         loadMasters();
 
@@ -122,4 +123,25 @@ public class DistributedMastershipStore
         return null;
     }
 
+    private class RemoteMasterShipEventHandler extends RemoteEventHandler<DeviceId, NodeId> {
+        public RemoteMasterShipEventHandler(LoadingCache<DeviceId, Optional<NodeId>> cache) {
+            super(cache);
+        }
+
+        @Override
+        protected void onAdd(DeviceId deviceId, NodeId nodeId) {
+            notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
+        }
+
+        @Override
+        protected void onRemove(DeviceId deviceId, NodeId nodeId) {
+            notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
+        }
+
+        @Override
+        protected void onUpdate(DeviceId deviceId, NodeId oldNodeId, NodeId nodeId) {
+            notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
+        }
+    }
+
 }

+package org.onlab.onos.store.flow.impl;
+
+import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_ADDED;
+import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_REMOVED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.ApplicationId;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.flow.DefaultFlowRule;
+import org.onlab.onos.net.flow.FlowRule;
+import org.onlab.onos.net.flow.FlowRule.FlowRuleState;
+import org.onlab.onos.net.flow.FlowRuleEvent;
+import org.onlab.onos.net.flow.FlowRuleEvent.Type;
+import org.onlab.onos.net.flow.FlowRuleStore;
+import org.onlab.onos.net.flow.FlowRuleStoreDelegate;
+import org.onlab.onos.store.AbstractStore;
+import org.slf4j.Logger;
+
+import com.google.common.collect.ArrayListMultimap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Multimap;
+
+/**
+ * Manages inventory of flow rules using trivial in-memory implementation.
+ */
+//FIXME: I LIE I AM NOT DISTRIBUTED
+@Component(immediate = true)
+@Service
+public class DistributedFlowRuleStore
+extends AbstractStore<FlowRuleEvent, FlowRuleStoreDelegate>
+implements FlowRuleStore {
+
+    private final Logger log = getLogger(getClass());
+
+    // store entries as a pile of rules, no info about device tables
+    private final Multimap<DeviceId, FlowRule> flowEntries =
+            ArrayListMultimap.<DeviceId, FlowRule>create();
+
+    private final Multimap<ApplicationId, FlowRule> flowEntriesById =
+            ArrayListMultimap.<ApplicationId, FlowRule>create();
+
+    @Activate
+    public void activate() {
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+
+
+    @Override
+    public synchronized FlowRule getFlowRule(FlowRule rule) {
+        for (FlowRule f : flowEntries.get(rule.deviceId())) {
+            if (f.equals(rule)) {
+                return f;
+            }
+        }
+        return null;
+    }
+
+    @Override
+    public synchronized Iterable<FlowRule> getFlowEntries(DeviceId deviceId) {
+        Collection<FlowRule> rules = flowEntries.get(deviceId);
+        if (rules == null) {
+            return Collections.emptyList();
+        }
+        return ImmutableSet.copyOf(rules);
+    }
+
+    @Override
+    public synchronized Iterable<FlowRule> getFlowEntriesByAppId(ApplicationId appId) {
+        Collection<FlowRule> rules = flowEntriesById.get(appId);
+        if (rules == null) {
+            return Collections.emptyList();
+        }
+        return ImmutableSet.copyOf(rules);
+    }
+
+    @Override
+    public synchronized void storeFlowRule(FlowRule rule) {
+        FlowRule f = new DefaultFlowRule(rule, FlowRuleState.PENDING_ADD);
+        DeviceId did = f.deviceId();
+        if (!flowEntries.containsEntry(did, f)) {
+            flowEntries.put(did, f);
+            flowEntriesById.put(rule.appId(), f);
+        }
+    }
+
+    @Override
+    public synchronized void deleteFlowRule(FlowRule rule) {
+        FlowRule f = new DefaultFlowRule(rule, FlowRuleState.PENDING_REMOVE);
+        DeviceId did = f.deviceId();
+
+        /*
+         *  find the rule and mark it for deletion.
+         *  Ultimately a flow removed will come remove it.
+         */
+
+        if (flowEntries.containsEntry(did, f)) {
+            //synchronized (flowEntries) {
+            flowEntries.remove(did, f);
+            flowEntries.put(did, f);
+            flowEntriesById.remove(rule.appId(), rule);
+            //}
+        }
+    }
+
+    @Override
+    public synchronized FlowRuleEvent addOrUpdateFlowRule(FlowRule rule) {
+        DeviceId did = rule.deviceId();
+
+        // check if this new rule is an update to an existing entry
+        if (flowEntries.containsEntry(did, rule)) {
+            //synchronized (flowEntries) {
+            // Multimaps support duplicates so we have to remove our rule
+            // and replace it with the current version.
+            flowEntries.remove(did, rule);
+            flowEntries.put(did, rule);
+            //}
+            return new FlowRuleEvent(Type.RULE_UPDATED, rule);
+        }
+
+        flowEntries.put(did, rule);
+        return new FlowRuleEvent(RULE_ADDED, rule);
+    }
+
+    @Override
+    public synchronized FlowRuleEvent removeFlowRule(FlowRule rule) {
+        //synchronized (this) {
+        if (flowEntries.remove(rule.deviceId(), rule)) {
+            return new FlowRuleEvent(RULE_REMOVED, rule);
+        } else {
+            return null;
+        }
+        //}
+    }
+
+
+
+
+
+
+
+}

+package org.onlab.onos.store.host.impl;
+
+import static org.onlab.onos.net.host.HostEvent.Type.HOST_ADDED;
+import static org.onlab.onos.net.host.HostEvent.Type.HOST_MOVED;
+import static org.onlab.onos.net.host.HostEvent.Type.HOST_REMOVED;
+import static org.onlab.onos.net.host.HostEvent.Type.HOST_UPDATED;
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DefaultHost;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Host;
+import org.onlab.onos.net.HostId;
+import org.onlab.onos.net.host.HostDescription;
+import org.onlab.onos.net.host.HostEvent;
+import org.onlab.onos.net.host.HostStore;
+import org.onlab.onos.net.host.HostStoreDelegate;
+import org.onlab.onos.net.host.PortAddresses;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.store.AbstractStore;
+import org.onlab.packet.IpPrefix;
+import org.onlab.packet.MacAddress;
+import org.onlab.packet.VlanId;
+import org.slf4j.Logger;
+
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Multimap;
+import com.google.common.collect.Sets;
+
+/**
+ * Manages inventory of end-station hosts using trivial in-memory
+ * implementation.
+ */
+//FIXME: I LIE I AM NOT DISTRIBUTED
+@Component(immediate = true)
+@Service
+public class DistributedHostStore
+extends AbstractStore<HostEvent, HostStoreDelegate>
+implements HostStore {
+
+    private final Logger log = getLogger(getClass());
+
+    // Host inventory
+    private final Map<HostId, Host> hosts = new ConcurrentHashMap<>();
+
+    // Hosts tracked by their location
+    private final Multimap<ConnectPoint, Host> locations = HashMultimap.create();
+
+    private final Map<ConnectPoint, PortAddresses> portAddresses =
+            new ConcurrentHashMap<>();
+
+    @Activate
+    public void activate() {
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+
+    @Override
+    public HostEvent createOrUpdateHost(ProviderId providerId, HostId hostId,
+            HostDescription hostDescription) {
+        Host host = hosts.get(hostId);
+        if (host == null) {
+            return createHost(providerId, hostId, hostDescription);
+        }
+        return updateHost(providerId, host, hostDescription);
+    }
+
+    // creates a new host and sends HOST_ADDED
+    private HostEvent createHost(ProviderId providerId, HostId hostId,
+            HostDescription descr) {
+        DefaultHost newhost = new DefaultHost(providerId, hostId,
+                descr.hwAddress(),
+                descr.vlan(),
+                descr.location(),
+                descr.ipAddresses());
+        synchronized (this) {
+            hosts.put(hostId, newhost);
+            locations.put(descr.location(), newhost);
+        }
+        return new HostEvent(HOST_ADDED, newhost);
+    }
+
+    // checks for type of update to host, sends appropriate event
+    private HostEvent updateHost(ProviderId providerId, Host host,
+            HostDescription descr) {
+        DefaultHost updated;
+        HostEvent event;
+        if (!host.location().equals(descr.location())) {
+            updated = new DefaultHost(providerId, host.id(),
+                    host.mac(),
+                    host.vlan(),
+                    descr.location(),
+                    host.ipAddresses());
+            event = new HostEvent(HOST_MOVED, updated);
+
+        } else if (!(host.ipAddresses().equals(descr.ipAddresses()))) {
+            updated = new DefaultHost(providerId, host.id(),
+                    host.mac(),
+                    host.vlan(),
+                    descr.location(),
+                    descr.ipAddresses());
+            event = new HostEvent(HOST_UPDATED, updated);
+        } else {
+            return null;
+        }
+        synchronized (this) {
+            hosts.put(host.id(), updated);
+            locations.remove(host.location(), host);
+            locations.put(updated.location(), updated);
+        }
+        return event;
+    }
+
+    @Override
+    public HostEvent removeHost(HostId hostId) {
+        synchronized (this) {
+            Host host = hosts.remove(hostId);
+            if (host != null) {
+                locations.remove((host.location()), host);
+                return new HostEvent(HOST_REMOVED, host);
+            }
+            return null;
+        }
+    }
+
+    @Override
+    public int getHostCount() {
+        return hosts.size();
+    }
+
+    @Override
+    public Iterable<Host> getHosts() {
+        return Collections.unmodifiableSet(new HashSet<>(hosts.values()));
+    }
+
+    @Override
+    public Host getHost(HostId hostId) {
+        return hosts.get(hostId);
+    }
+
+    @Override
+    public Set<Host> getHosts(VlanId vlanId) {
+        Set<Host> vlanset = new HashSet<>();
+        for (Host h : hosts.values()) {
+            if (h.vlan().equals(vlanId)) {
+                vlanset.add(h);
+            }
+        }
+        return vlanset;
+    }
+
+    @Override
+    public Set<Host> getHosts(MacAddress mac) {
+        Set<Host> macset = new HashSet<>();
+        for (Host h : hosts.values()) {
+            if (h.mac().equals(mac)) {
+                macset.add(h);
+            }
+        }
+        return macset;
+    }
+
+    @Override
+    public Set<Host> getHosts(IpPrefix ip) {
+        Set<Host> ipset = new HashSet<>();
+        for (Host h : hosts.values()) {
+            if (h.ipAddresses().contains(ip)) {
+                ipset.add(h);
+            }
+        }
+        return ipset;
+    }
+
+    @Override
+    public Set<Host> getConnectedHosts(ConnectPoint connectPoint) {
+        return ImmutableSet.copyOf(locations.get(connectPoint));
+    }
+
+    @Override
+    public Set<Host> getConnectedHosts(DeviceId deviceId) {
+        Set<Host> hostset = new HashSet<>();
+        for (ConnectPoint p : locations.keySet()) {
+            if (p.deviceId().equals(deviceId)) {
+                hostset.addAll(locations.get(p));
+            }
+        }
+        return hostset;
+    }
+
+    @Override
+    public void updateAddressBindings(PortAddresses addresses) {
+        synchronized (portAddresses) {
+            PortAddresses existing = portAddresses.get(addresses.connectPoint());
+            if (existing == null) {
+                portAddresses.put(addresses.connectPoint(), addresses);
+            } else {
+                Set<IpPrefix> union = Sets.union(existing.ips(), addresses.ips())
+                        .immutableCopy();
+
+                MacAddress newMac = (addresses.mac() == null) ? existing.mac()
+                        : addresses.mac();
+
+                PortAddresses newAddresses =
+                        new PortAddresses(addresses.connectPoint(), union, newMac);
+
+                portAddresses.put(newAddresses.connectPoint(), newAddresses);
+            }
+        }
+    }
+
+    @Override
+    public void removeAddressBindings(PortAddresses addresses) {
+        synchronized (portAddresses) {
+            PortAddresses existing = portAddresses.get(addresses.connectPoint());
+            if (existing != null) {
+                Set<IpPrefix> difference =
+                        Sets.difference(existing.ips(), addresses.ips()).immutableCopy();
+
+                // If they removed the existing mac, set the new mac to null.
+                // Otherwise, keep the existing mac.
+                MacAddress newMac = existing.mac();
+                if (addresses.mac() != null && addresses.mac().equals(existing.mac())) {
+                    newMac = null;
+                }
+
+                PortAddresses newAddresses =
+                        new PortAddresses(addresses.connectPoint(), difference, newMac);
+
+                portAddresses.put(newAddresses.connectPoint(), newAddresses);
+            }
+        }
+    }
+
+    @Override
+    public void clearAddressBindings(ConnectPoint connectPoint) {
+        synchronized (portAddresses) {
+            portAddresses.remove(connectPoint);
+        }
+    }
+
+    @Override
+    public Set<PortAddresses> getAddressBindings() {
+        synchronized (portAddresses) {
+            return new HashSet<>(portAddresses.values());
+        }
+    }
+
+    @Override
+    public PortAddresses getAddressBindingsForPort(ConnectPoint connectPoint) {
+        PortAddresses addresses;
+
+        synchronized (portAddresses) {
+            addresses = portAddresses.get(connectPoint);
+        }
+
+        if (addresses == null) {
+            addresses = new PortAddresses(connectPoint, null, null);
+        }
+
+        return addresses;
+    }
+
+}

+package org.onlab.onos.store.topology.impl;
+
+import org.onlab.graph.AdjacencyListsGraph;
+import org.onlab.onos.net.topology.TopologyEdge;
+import org.onlab.onos.net.topology.TopologyGraph;
+import org.onlab.onos.net.topology.TopologyVertex;
+
+import java.util.Set;
+
+/**
+ * Default implementation of an immutable topology graph based on a generic
+ * implementation of adjacency lists graph.
+ */
+public class DefaultTopologyGraph
+        extends AdjacencyListsGraph<TopologyVertex, TopologyEdge>
+        implements TopologyGraph {
+
+    /**
+     * Creates a topology graph comprising of the specified vertexes and edges.
+     *
+     * @param vertexes set of graph vertexes
+     * @param edges    set of graph edges
+     */
+    public DefaultTopologyGraph(Set<TopologyVertex> vertexes, Set<TopologyEdge> edges) {
+        super(vertexes, edges);
+    }
+
+}

+package org.onlab.onos.store.topology.impl;
+
+import static org.slf4j.LoggerFactory.getLogger;
+
+import java.util.List;
+import java.util.Set;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.event.Event;
+import org.onlab.onos.net.ConnectPoint;
+import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.Link;
+import org.onlab.onos.net.Path;
+import org.onlab.onos.net.provider.ProviderId;
+import org.onlab.onos.net.topology.ClusterId;
+import org.onlab.onos.net.topology.GraphDescription;
+import org.onlab.onos.net.topology.LinkWeight;
+import org.onlab.onos.net.topology.Topology;
+import org.onlab.onos.net.topology.TopologyCluster;
+import org.onlab.onos.net.topology.TopologyEvent;
+import org.onlab.onos.net.topology.TopologyGraph;
+import org.onlab.onos.net.topology.TopologyStore;
+import org.onlab.onos.net.topology.TopologyStoreDelegate;
+import org.onlab.onos.store.AbstractStore;
+import org.slf4j.Logger;
+
+/**
+ * Manages inventory of topology snapshots using trivial in-memory
+ * structures implementation.
+ */
+//FIXME: I LIE I AM NOT DISTRIBUTED
+@Component(immediate = true)
+@Service
+public class DistributedTopologyStore
+extends AbstractStore<TopologyEvent, TopologyStoreDelegate>
+implements TopologyStore {
+
+    private final Logger log = getLogger(getClass());
+
+    private volatile DefaultTopology current;
+
+    @Activate
+    public void activate() {
+        log.info("Started");
+    }
+
+    @Deactivate
+    public void deactivate() {
+        log.info("Stopped");
+    }
+    @Override
+    public Topology currentTopology() {
+        return current;
+    }
+
+    @Override
+    public boolean isLatest(Topology topology) {
+        // Topology is current only if it is the same as our current topology
+        return topology == current;
+    }
+
+    @Override
+    public TopologyGraph getGraph(Topology topology) {
+        return defaultTopology(topology).getGraph();
+    }
+
+    @Override
+    public Set<TopologyCluster> getClusters(Topology topology) {
+        return defaultTopology(topology).getClusters();
+    }
+
+    @Override
+    public TopologyCluster getCluster(Topology topology, ClusterId clusterId) {
+        return defaultTopology(topology).getCluster(clusterId);
+    }
+
+    @Override
+    public Set<DeviceId> getClusterDevices(Topology topology, TopologyCluster cluster) {
+        return defaultTopology(topology).getClusterDevices(cluster);
+    }
+
+    @Override
+    public Set<Link> getClusterLinks(Topology topology, TopologyCluster cluster) {
+        return defaultTopology(topology).getClusterLinks(cluster);
+    }
+
+    @Override
+    public Set<Path> getPaths(Topology topology, DeviceId src, DeviceId dst) {
+        return defaultTopology(topology).getPaths(src, dst);
+    }
+
+    @Override
+    public Set<Path> getPaths(Topology topology, DeviceId src, DeviceId dst,
+            LinkWeight weight) {
+        return defaultTopology(topology).getPaths(src, dst, weight);
+    }
+
+    @Override
+    public boolean isInfrastructure(Topology topology, ConnectPoint connectPoint) {
+        return defaultTopology(topology).isInfrastructure(connectPoint);
+    }
+
+    @Override
+    public boolean isBroadcastPoint(Topology topology, ConnectPoint connectPoint) {
+        return defaultTopology(topology).isBroadcastPoint(connectPoint);
+    }
+
+    @Override
+    public TopologyEvent updateTopology(ProviderId providerId,
+            GraphDescription graphDescription,
+            List<Event> reasons) {
+        // First off, make sure that what we're given is indeed newer than
+        // what we already have.
+        if (current != null && graphDescription.timestamp() < current.time()) {
+            return null;
+        }
+
+        // Have the default topology construct self from the description data.
+        DefaultTopology newTopology =
+                new DefaultTopology(providerId, graphDescription);
+
+        // Promote the new topology to current and return a ready-to-send event.
+        synchronized (this) {
+            current = newTopology;
+            return new TopologyEvent(TopologyEvent.Type.TOPOLOGY_CHANGED, current);
+        }
+    }
+
+    // Validates the specified topology and returns it as a default
+    private DefaultTopology defaultTopology(Topology topology) {
+        if (topology instanceof DefaultTopology) {
+            return (DefaultTopology) topology;
+        }
+        throw new IllegalArgumentException("Topology class " + topology.getClass() +
+                " not supported");
+    }
+
+}

+package org.onlab.onos.store.topology.impl;
+
+import org.onlab.onos.net.DeviceId;
+
+import java.util.Objects;
+
+/**
+ * Key for filing pre-computed paths between source and destination devices.
+ */
+class PathKey {
+    private final DeviceId src;
+    private final DeviceId dst;
+
+    /**
+     * Creates a path key from the given source/dest pair.
+     * @param src source device
+     * @param dst destination device
+     */
+    PathKey(DeviceId src, DeviceId dst) {
+        this.src = src;
+        this.dst = dst;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(src, dst);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+        if (obj instanceof PathKey) {
+            final PathKey other = (PathKey) obj;
+            return Objects.equals(this.src, other.src) && Objects.equals(this.dst, other.dst);
+        }
+        return false;
+    }
+}

@@ -93,12 +93,16 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
 
     @Override
     public final void sendMsg(OFMessage m) {
-        this.write(m);
+        if (role == RoleState.MASTER) {
+            this.write(m);
+        }
     }
 
     @Override
     public final void sendMsg(List<OFMessage> msgs) {
-        this.write(msgs);
+        if (role == RoleState.MASTER) {
+            this.write(msgs);
+        }
     }
 
     @Override
@@ -164,7 +168,9 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
      */
     @Override
     public final void handleMessage(OFMessage m) {
-        this.agent.processMessage(dpid, m);
+        if (this.role == RoleState.MASTER) {
+            this.agent.processMessage(dpid, m);
+        }
     }
 
     @Override
@@ -226,19 +232,34 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
     @Override
     public abstract void processDriverHandshakeMessage(OFMessage m);
 
+
+    // Role Handling
+
     @Override
     public void setRole(RoleState role) {
         try {
-            log.info("Sending role {} to switch {}", role, getStringId());
             if (this.roleMan.sendRoleRequest(role, RoleRecvStatus.MATCHED_SET_ROLE)) {
-                this.role = role;
+                log.info("Sending role {} to switch {}", role, getStringId());
+                if (role == RoleState.SLAVE || role == RoleState.EQUAL) {
+                    this.role = role;
+                }
             }
         } catch (IOException e) {
             log.error("Unable to write to switch {}.", this.dpid);
         }
     }
 
-    // Role Handling
+    @Override
+    public void reassertRole() {
+        if (this.getRole() == RoleState.MASTER) {
+            log.warn("Received permission error from switch {} while " +
+                    "being master. Reasserting master role.",
+                    this.getStringId());
+            this.setRole(RoleState.MASTER);
+        }
+    }
+
+
 
     @Override
     public void handleRole(OFMessage m) throws SwitchStateException {
@@ -246,11 +267,15 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
         RoleRecvStatus rrs = roleMan.deliverRoleReply(rri);
         if (rrs == RoleRecvStatus.MATCHED_SET_ROLE) {
             if (rri.getRole() == RoleState.MASTER) {
+                this.role = rri.getRole();
                 this.transitionToMasterSwitch();
             } else if (rri.getRole() == RoleState.EQUAL ||
-                    rri.getRole() == RoleState.MASTER) {
+                    rri.getRole() == RoleState.SLAVE) {
                 this.transitionToEqualSwitch();
             }
+        }  else {
+            return;
+            //TODO: tell people that we failed.
         }
     }
 
@@ -267,11 +292,15 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
                 new RoleReplyInfo(r, null, m.getXid()));
         if (rrs == RoleRecvStatus.MATCHED_SET_ROLE) {
             if (r == RoleState.MASTER) {
+                this.role = r;
                 this.transitionToMasterSwitch();
             } else if (r == RoleState.EQUAL ||
                     r == RoleState.SLAVE) {
                 this.transitionToEqualSwitch();
             }
+        } else {
+            return;
+            //TODO: tell people that we failed.
         }
     }
 
@@ -285,12 +314,7 @@ public abstract class AbstractOpenFlowSwitch implements OpenFlowSwitchDriver {
         return true;
     }
 
-    @Override
-    public void reassertRole() {
-        if (this.getRole() == RoleState.MASTER) {
-            this.setRole(RoleState.MASTER);
-        }
-    }
+
 
     @Override
     public final void setAgent(OpenFlowAgent ag) {

@@ -521,9 +521,7 @@ class OFChannelHandler extends IdleStateAwareChannelHandler {
                     // if two controllers are master (even if its only for
                     // a brief period). We might need to see if these errors
                     // persist before we reassert
-                    log.warn("Received permission error from switch {} while" +
-                            "being master. Reasserting master role.",
-                            h.getSwitchInfoString());
+
                     h.sw.reassertRole();
                 } else if (m.getErrType() == OFErrorType.FLOW_MOD_FAILED &&
                         ((OFFlowModFailedErrorMsg) m).getCode() ==

@@ -142,9 +142,9 @@ class RoleManager implements RoleHandler {
             }
             // OF1.0 switch with support for NX_ROLE_REQUEST vendor extn.
             // make Role.EQUAL become Role.SLAVE
+            pendingRole = role;
             role = (role == RoleState.EQUAL) ? RoleState.SLAVE : role;
             pendingXid = sendNxRoleRequest(role);
-            pendingRole = role;
             requestPending = true;
         } else {
             // OF1.3 switch, use OFPT_ROLE_REQUEST message

 #!/usr/bin/python
 import sys, solar
-topo = solar.Solar(cip=sys.argv[1])
+topo = solar.Solar(cips=sys.argv[1:])
 topo.run()

@@ -17,22 +17,22 @@ class CustomCLI(CLI):
 class Solar(object):
     """ Create a tiered topology from semi-scratch in Mininet """
 
-    def __init__(self, cname='onos', cip='192.168.56.1', islands=3, edges=2, hosts=2,
-                 proto=None):
+    def __init__(self, cname='onos', cips=['192.168.56.1'], islands=3, edges=2, hosts=2):
         """Create tower topology for mininet"""
 
         # We are creating the controller with local-loopback on purpose to avoid
         # having the switches connect immediately. Instead, we'll set controller
         # explicitly for each switch after configuring it as we want.
-        self.flare = RemoteController(cname, cip, 6633)
-        self.net = Mininet(controller=self.flare, switch = OVSKernelSwitch,
+        self.ctrls = [ RemoteController(cname, cip, 6633) for cip in cips ]
+        self.net = Mininet(controller=RemoteController, switch = OVSKernelSwitch,
                            build=False)
 
-        self.cip = cip
+        self.cips = cips
         self.spines = []
         self.leaves = []
         self.hosts = []
-        self.proto = proto
+        for ctrl in self.ctrls:
+            self.net.addController(ctrl)
 
         # Create the two core switches and links between them
         c1 = self.net.addSwitch('c1',dpid='1111000000000000')
@@ -83,29 +83,11 @@ class Solar(object):
 
     def run(self):
         """ Runs the created network topology and launches mininet cli"""
-        self.run_silent()
+        self.net.build()
+        self.net.start()
         CustomCLI(self.net)
         self.net.stop()
 
-    def run_silent(self):
-        """ Runs silently - for unit testing """
-        self.net.build()
-
-        # Start the switches, configure them with desired protocols and only
-        # then set the controller
-        for sw in self.spines:
-            sw.start([self.flare])
-            if self.proto:
-                sw.cmd('ovs-vsctl set bridge %(sw)s protocols=%(proto)s' % \
-                           { 'sw': sw.name, 'proto': self.proto})
-            sw.cmdPrint('ovs-vsctl set-controller %(sw)s tcp:%(ctl)s:6633' % \
-                            {'sw': sw.name, 'ctl': self.cip})
-
-        for sw in self.leaves:
-            sw.start([self.flare])
-            sw.cmdPrint('ovs-vsctl set-controller %(sw)s tcp:%(ctl)s:6633' % \
-                            {'sw': sw.name, 'ctl': self.cip})
-
     def pingAll(self):
         """ PingAll to create flows - for unit testing """
         self.net.pingAll()