MasterServer.java

Go to the documentation of this file.

/*
 * Copyright (C) 2011 Google Inc.
 * 
 * Licensed 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.ros.internal.node.server.master;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.Lists;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.ros.address.AdvertiseAddress;
import org.ros.address.BindAddress;
import org.ros.internal.node.client.SlaveClient;
import org.ros.internal.node.server.NodeIdentifier;
import org.ros.internal.node.server.SlaveServer;
import org.ros.internal.node.server.XmlRpcServer;
import org.ros.internal.node.topic.TopicParticipant;
import org.ros.internal.node.xmlrpc.MasterXmlRpcEndpointImpl;
import org.ros.master.client.TopicSystemState;
import org.ros.namespace.GraphName;
import org.ros.node.Node;
import org.ros.node.service.ServiceServer;
import org.ros.node.topic.Publisher;
import org.ros.node.topic.Subscriber;

import java.net.URI;
import java.util.Collection;
import java.util.List;

public class MasterServer extends XmlRpcServer implements MasterRegistrationListener {

  private static final boolean DEBUG = false;
  private static final Log log = LogFactory.getLog(MasterServer.class);

  public static final int SYSTEM_STATE_PUBLISHERS = 0;

  public static final int SYSTEM_STATE_SUBSCRIBERS = 1;

  public static final int SYSTEM_STATE_SERVICES = 2;

  private static final GraphName MASTER_NODE_NAME = GraphName.of("/master");

  private final MasterRegistrationManagerImpl masterRegistrationManager;

  public MasterServer(BindAddress bindAddress, AdvertiseAddress advertiseAddress) {
    super(bindAddress, advertiseAddress);
    masterRegistrationManager = new MasterRegistrationManagerImpl(this);
  }

  public void start() {
    if (DEBUG) {
      log.info("Starting master server.");
    }
    super.start(MasterXmlRpcEndpointImpl.class, new MasterXmlRpcEndpointImpl(this));
  }

  public void registerService(GraphName nodeName, URI nodeSlaveUri, GraphName serviceName,
      URI serviceUri) {
    synchronized (masterRegistrationManager) {
      masterRegistrationManager.registerService(nodeName, nodeSlaveUri, serviceName, serviceUri);
    }
  }

  public boolean unregisterService(GraphName nodeName, GraphName serviceName, URI serviceUri) {
    synchronized (masterRegistrationManager) {
      return masterRegistrationManager.unregisterService(nodeName, serviceName, serviceUri);
    }
  }

  public List<URI> registerSubscriber(GraphName nodeName, URI nodeSlaveUri, GraphName topicName,
      String topicMessageType) {
    if (DEBUG) {
      log.info(String.format(
          "Registering subscriber %s with message type %s on node %s with URI %s", topicName,
          topicMessageType, nodeName, nodeSlaveUri));
    }

    synchronized (masterRegistrationManager) {
      TopicRegistrationInfo topicInfo =
          masterRegistrationManager.registerSubscriber(nodeName, nodeSlaveUri, topicName,
              topicMessageType);
      List<URI> publisherUris = Lists.newArrayList();
      for (NodeRegistrationInfo publisherNodeInfo : topicInfo.getPublishers()) {
        publisherUris.add(publisherNodeInfo.getNodeSlaveUri());
      }
      return publisherUris;
    }
  }

  public boolean unregisterSubscriber(GraphName nodeName, GraphName topicName) {
    if (DEBUG) {
      log.info(String.format("Unregistering subscriber for %s on node %s.", topicName, nodeName));
    }
    synchronized (masterRegistrationManager) {
      return masterRegistrationManager.unregisterSubscriber(nodeName, topicName);
    }
  }

  public List<URI> registerPublisher(GraphName nodeName, URI nodeSlaveUri, GraphName topicName,
      String topicMessageType) {
    if (DEBUG) {
      log.info(String.format(
          "Registering publisher %s with message type %s on node %s with URI %s.", topicName,
          topicMessageType, nodeName, nodeSlaveUri));
    }

    synchronized (masterRegistrationManager) {
      TopicRegistrationInfo topicInfo =
          masterRegistrationManager.registerPublisher(nodeName, nodeSlaveUri, topicName,
              topicMessageType);

      List<URI> subscriberSlaveUris = Lists.newArrayList();
      for (NodeRegistrationInfo publisherNodeInfo : topicInfo.getSubscribers()) {
        subscriberSlaveUris.add(publisherNodeInfo.getNodeSlaveUri());
      }

      publisherUpdate(topicInfo, subscriberSlaveUris);

      return subscriberSlaveUris;
    }
  }

  private void publisherUpdate(TopicRegistrationInfo topicInfo, List<URI> subscriberSlaveUris) {
    if (DEBUG) {
      log.info("Publisher update: " + topicInfo.getTopicName());
    }
    List<URI> publisherUris = Lists.newArrayList();
    for (NodeRegistrationInfo publisherNodeInfo : topicInfo.getPublishers()) {
      publisherUris.add(publisherNodeInfo.getNodeSlaveUri());
    }

    GraphName topicName = topicInfo.getTopicName();
    for (URI subscriberSlaveUri : subscriberSlaveUris) {
      contactSubscriberForPublisherUpdate(subscriberSlaveUri, topicName, publisherUris);
    }
  }

  @VisibleForTesting
  protected void contactSubscriberForPublisherUpdate(URI subscriberSlaveUri, GraphName topicName,
      List<URI> publisherUris) {
    SlaveClient client = new SlaveClient(MASTER_NODE_NAME, subscriberSlaveUri);
    client.publisherUpdate(topicName, publisherUris);
  }

  public boolean unregisterPublisher(GraphName nodeName, GraphName topicName) {
    if (DEBUG) {
      log.info(String.format("Unregistering publisher for %s on %s.", topicName, nodeName));
    }
    synchronized (masterRegistrationManager) {
      return masterRegistrationManager.unregisterPublisher(nodeName, topicName);
    }
  }

  public URI lookupNode(GraphName nodeName) {
    synchronized (masterRegistrationManager) {
      NodeRegistrationInfo node = masterRegistrationManager.getNodeRegistrationInfo(nodeName);
      if (node != null) {
        return node.getNodeSlaveUri();
      } else {
        return null;
      }
    }
  }

  public List<List<String>> getTopicTypes(GraphName calledId) {
    synchronized (masterRegistrationManager) {
      List<List<String>> result = Lists.newArrayList();
      for (TopicRegistrationInfo topic : masterRegistrationManager.getAllTopics()) {
        result.add(Lists.newArrayList(topic.getTopicName().toString(), topic.getMessageType()));
      }
      return result;
    }
  }

  public List<Object> getSystemState() {
    synchronized (masterRegistrationManager) {
      List<Object> result = Lists.newArrayList();

      Collection<TopicRegistrationInfo> topics = masterRegistrationManager.getAllTopics();
      result.add(getSystemStatePublishers(topics));
      result.add(getSystemStateSubscribers(topics));
      result.add(getSystemStateServices());
      return result;
    }
  }

  private List<Object> getSystemStatePublishers(Collection<TopicRegistrationInfo> topics) {
    List<Object> result = Lists.newArrayList();
    for (TopicRegistrationInfo topic : topics) {
      if (topic.hasPublishers()) {
        List<Object> topicInfo = Lists.newArrayList();
        topicInfo.add(topic.getTopicName().toString());

        List<String> publist = Lists.newArrayList();
        for (NodeRegistrationInfo node : topic.getPublishers()) {
          publist.add(node.getNodeName().toString());
        }
        topicInfo.add(publist);

        result.add(topicInfo);
      }
    }
    return result;
  }

  private List<Object> getSystemStateSubscribers(Collection<TopicRegistrationInfo> topics) {
    List<Object> result = Lists.newArrayList();
    for (TopicRegistrationInfo topic : topics) {
      if (topic.hasSubscribers()) {
        List<Object> topicInfo = Lists.newArrayList();
        topicInfo.add(topic.getTopicName().toString());

        List<Object> sublist = Lists.newArrayList();
        for (NodeRegistrationInfo node : topic.getSubscribers()) {
          sublist.add(node.getNodeName().toString());
        }
        topicInfo.add(sublist);

        result.add(topicInfo);
      }
    }
    return result;
  }

  private List<Object> getSystemStateServices() {
    List<Object> result = Lists.newArrayList();

    for (ServiceRegistrationInfo service : masterRegistrationManager.getAllServices()) {
      List<Object> topicInfo = Lists.newArrayList();
      topicInfo.add(service.getServiceName().toString());
      topicInfo.add(Lists.newArrayList(service.getServiceName().toString()));

      result.add(topicInfo);
    }

    return result;
  }

  public URI lookupService(GraphName serviceName) {
    synchronized (masterRegistrationManager) {
      ServiceRegistrationInfo service =
          masterRegistrationManager.getServiceRegistrationInfo(serviceName);
      if (service != null) {
        return service.getServiceUri();
      } else {
        return null;
      }
    }
  }

  public List<Object> getPublishedTopics(GraphName caller, GraphName subgraph) {
    synchronized (masterRegistrationManager) {
      // TODO(keith): Filter topics according to subgraph.
      List<Object> result = Lists.newArrayList();
      for (TopicRegistrationInfo topic : masterRegistrationManager.getAllTopics()) {
        if (topic.hasPublishers()) {
          result.add(Lists.newArrayList(topic.getTopicName().toString(), topic.getMessageType()));
        }
      }
      return result;
    }
  }

  @Override
  public void onNodeReplacement(NodeRegistrationInfo nodeInfo) {
    // A node in the registration manager is being replaced. Contact the node
    // and tell it to shut down.
    if (log.isWarnEnabled()) {
      log.warn(String.format("Existing node %s with slave URI %s will be shutdown.",
          nodeInfo.getNodeName(), nodeInfo.getNodeSlaveUri()));
    }

    SlaveClient client = new SlaveClient(MASTER_NODE_NAME, nodeInfo.getNodeSlaveUri());
    client.shutdown("Replaced by new slave");
  }
}