Process dispatch

3rdparty/libprocess/include/process/dispatch.hpp

@@ -347,6 +347,33 @@ Future<R> dispatch(
 #undef TEMPLATE
 
 
+template <typename I, typename Callable, typename ...Args>
+auto dispatch(
+    const ProcessBase* process,
+    Callable fn,
+    Args&&... args) -> typename std::result_of<decltype(fn)(I*, Args...)>::type
+{
+  typedef typename std::result_of<decltype(fn)(I*, Args...)>::type::value_type result_type; //NOLINT
+  std::shared_ptr<Promise<result_type>> promise(new Promise<result_type>());
+
+  auto call(std::bind(fn, std::placeholders::_1, std::forward<Args>(args)...));
+
+  std::shared_ptr<std::function<void(ProcessBase*)>> f(
+      new std::function<void(ProcessBase*)>(
+      [=](ProcessBase* process) {
+        assert(process != NULL);
+        I* t = dynamic_cast<I*>(process);
+        assert(t != NULL);
+
+        promise->associate(call(t));
+      }));
+
+  internal::dispatch(process->self(), f, &typeid(fn));
+
+  return promise->future();
+}
+
+
 inline void dispatch(
     const UPID& pid,
     const std::function<void()>& f)

3rdparty/libprocess/include/process/future.hpp

@@ -90,6 +90,8 @@ template <typename T>
 class Future
 {
 public:
+  typedef T value_type;
+
   // Constructs a failed future.
   static Future<T> failed(const std::string& message);
 

src/Makefile.am

@@ -706,6 +706,7 @@ libmesos_no_3rdparty_la_SOURCES +=					\
 	common/date_utils.hpp						\
 	common/http.hpp							\
 	common/parse.hpp						\
+	common/process_dispatcher.hpp					\
 	common/protobuf_utils.hpp					\
 	common/recordio.hpp						\
 	common/resources_utils.hpp					\

src/common/process_dispatcher.hpp

@@ -0,0 +1,117 @@
+/**
+ * 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.
+ */
+
+#ifndef __PROCESS_DISPATCHER_HPP__
+#define __PROCESS_DISPATCHER_HPP__
+
+#include <memory>
+#include <utility>
+
+#include <stout/try.hpp>
+
+#include <process/dispatch.hpp>
+#include <process/process.hpp>
+#include <process/shared.hpp>
+
+namespace mesos {
+
+template <typename I>
+class Dispatchable
+{
+public:
+  virtual ~Dispatchable(){}
+
+  template <typename Callable, typename ...Args>
+    auto dispatch(Callable fn, Args&&... args)
+    -> typename std::result_of<decltype(fn)(I*, Args...)>::type
+    {
+      ProcessBase* process = dynamic_cast<ProcessBase*>(getInterface());
+      return process::dispatch<I, Callable, Args...>(
+          process,
+          fn,
+          std::forward<Args>(args)...);
+    }
+
+protected:
+  void setInterface(const process::Shared<I>& i)
+  {
+    interface = i;
+  }
+
+  I* getInterface()
+  {
+    return const_cast<I*>(interface.get());
+  }
+
+  process::Shared<I> interface;
+};
+
+template <typename I, typename P = I>
+class ProcessDispatcher : public Dispatchable<I>
+{
+public:
+  template<typename ...Args>
+  static Try<process::Owned<Dispatchable<I>>> create(Args&&... args)
+  {
+    Try<process::Owned<P>> newProcess(P::create(std::forward<Args>(args)...));
+
+    if (newProcess.isError()) {
+      return Error(newProcess.error());
+    }
+
+    process::Shared<I> sharedInterface(newProcess.get().release());
+
+    return process::Owned<Dispatchable<I>>(
+        new ProcessDispatcher(sharedInterface));
+  }
+
+  static Try<process::Owned<ProcessDispatcher>> create(
+      process::Shared<I> process)
+  {
+    return(process::Owned<ProcessDispatcher>(new ProcessDispatcher(process)));
+  }
+
+  ~ProcessDispatcher()
+  {
+    ProcessBase* process = dynamic_cast<ProcessBase*>(this->getInterface());
+
+    terminate(process);
+    process::wait(process);
+  }
+
+private:
+  ProcessDispatcher(const process::Shared<I>& i)
+    :Dispatchable<I>()
+  {
+    this->setInterface(i);
+
+    ProcessBase* process = dynamic_cast<ProcessBase*>(const_cast<I*>(i.get()));
+    if (!process)
+    {
+      assert(false);
+    }
+
+    spawn(process);
+  }
+
+  ProcessDispatcher(const ProcessDispatcher&) = delete;
+};
+
+} // namespace mesos {
+
+#endif // __PROCESS_DISPATCHER_HPP__

src/slave/containerizer/provisioners/docker/token_manager.cpp

@@ -42,68 +42,6 @@ namespace slave {
 namespace docker {
 namespace registry {
 
-class TokenManagerProcess : public Process<TokenManagerProcess>
-{
-public:
-  static Try<Owned<TokenManagerProcess>> create(const URL& realm);
-
-  Future<Token> getToken(
-      const string& service,
-      const string& scope,
-      const Option<string>& account);
-
-private:
-  static const string TOKEN_PATH_PREFIX;
-  static const Duration RESPONSE_TIMEOUT;
-
-  TokenManagerProcess(const URL& realm)
-    : realm_(realm) {}
-
-  Try<Token> getTokenFromResponse(const Response& response) const;
-
-  /**
-   * Key for the token cache.
-   */
-  struct TokenCacheKey
-  {
-    string service;
-    string scope;
-  };
-
-  struct TokenCacheKeyHash
-  {
-    size_t operator()(const TokenCacheKey& key) const
-    {
-      hash<string> hashFn;
-
-      return (hashFn(key.service) ^
-          (hashFn(key.scope) << 1));
-    }
-  };
-
-  struct TokenCacheKeyEqual
-  {
-    bool operator()(
-        const TokenCacheKey& left,
-        const TokenCacheKey& right) const
-    {
-      return ((left.service == right.service) &&
-          (left.scope == right.scope));
-    }
-  };
-
-  typedef hashmap<
-    const TokenCacheKey,
-    Token,
-    TokenCacheKeyHash,
-    TokenCacheKeyEqual> TokenCacheType;
-
-  const URL realm_;
-  TokenCacheType tokenCache_;
-
-  TokenManagerProcess(const TokenManagerProcess&) = delete;
-  TokenManagerProcess& operator=(const TokenManagerProcess&) = delete;
-};
 
 const Duration TokenManagerProcess::RESPONSE_TIMEOUT = Seconds(10);
 const string TokenManagerProcess::TOKEN_PATH_PREFIX = "/v2/token/";
@@ -233,46 +171,6 @@ bool Token::isValid() const
 }
 
 
-Try<Owned<TokenManager>> TokenManager::create(
-    const URL& realm)
-{
-  Try<Owned<TokenManagerProcess>> process = TokenManagerProcess::create(realm);
-  if (process.isError()) {
-    return Error(process.error());
-  }
-
-  return Owned<TokenManager>(new TokenManager(process.get()));
-}
-
-
-TokenManager::TokenManager(Owned<TokenManagerProcess>& process)
-  : process_(process)
-{
-  spawn(CHECK_NOTNULL(process_.get()));
-}
-
-
-TokenManager::~TokenManager()
-{
-  terminate(process_.get());
-  process::wait(process_.get());
-}
-
-
-Future<Token> TokenManager::getToken(
-    const string& service,
-    const string& scope,
-    const Option<string>& account)
-{
-  return dispatch(
-      process_.get(),
-      &TokenManagerProcess::getToken,
-      service,
-      scope,
-      account);
-}
-
-
 Try<Owned<TokenManagerProcess>> TokenManagerProcess::create(const URL& realm)
 {
   return Owned<TokenManagerProcess>(new TokenManagerProcess(realm));
@@ -308,6 +206,7 @@ Future<Token> TokenManagerProcess::getToken(
     const string& scope,
     const Option<string>& account)
 {
+        std::cout << __FILE__ << ":" << __LINE__ << std::endl;
   const TokenCacheKey tokenKey = {service, scope};
 
   if (tokenCache_.contains(tokenKey)) {