add driver_cloud_swap_server for remote vtd.

include/common.pri

@@ -6,7 +6,10 @@ unix:DEFINES += RUNSYSTEMTEST
 if(contains(DEFINES,RUNSYSTEMTEST)){
 unix:system("./linuxsystemtest.sh ")
 }
-DEFINES =+ MODULECOMM_NO_FASTRTPS
+
+DEFINES += MODULECOMM_NO_FASTRTPS
+
+
 unix:include(systemdef.pri)
 win32: DEFINES += SYSTEM_WIN
 

src/driver/driver_cloud_swap_server/cloudswap.grpc.pb.cc

@@ -0,0 +1,78 @@
+// Generated by the gRPC C++ plugin.
+// If you make any local change, they will be lost.
+// source: cloudswap.proto
+
+#include "cloudswap.pb.h"
+#include "cloudswap.grpc.pb.h"
+
+#include <functional>
+#include <grpcpp/impl/codegen/async_stream.h>
+#include <grpcpp/impl/codegen/async_unary_call.h>
+#include <grpcpp/impl/codegen/channel_interface.h>
+#include <grpcpp/impl/codegen/client_unary_call.h>
+#include <grpcpp/impl/codegen/client_callback.h>
+#include <grpcpp/impl/codegen/message_allocator.h>
+#include <grpcpp/impl/codegen/method_handler.h>
+#include <grpcpp/impl/codegen/rpc_service_method.h>
+#include <grpcpp/impl/codegen/server_callback.h>
+#include <grpcpp/impl/codegen/server_callback_handlers.h>
+#include <grpcpp/impl/codegen/server_context.h>
+#include <grpcpp/impl/codegen/service_type.h>
+#include <grpcpp/impl/codegen/sync_stream.h>
+namespace iv {
+
+static const char* CloudSwapStream_method_names[] = {
+  "/iv.CloudSwapStream/swap",
+};
+
+std::unique_ptr< CloudSwapStream::Stub> CloudSwapStream::NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options) {
+  (void)options;
+  std::unique_ptr< CloudSwapStream::Stub> stub(new CloudSwapStream::Stub(channel));
+  return stub;
+}
+
+CloudSwapStream::Stub::Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel)
+  : channel_(channel), rpcmethod_swap_(CloudSwapStream_method_names[0], ::grpc::internal::RpcMethod::BIDI_STREAMING, channel)
+  {}
+
+::grpc::ClientReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* CloudSwapStream::Stub::swapRaw(::grpc::ClientContext* context) {
+  return ::grpc::internal::ClientReaderWriterFactory< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>::Create(channel_.get(), rpcmethod_swap_, context);
+}
+
+void CloudSwapStream::Stub::experimental_async::swap(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor< ::iv::CloudSwapRequestStream,::iv::CloudSwapReplyStream>* reactor) {
+  ::grpc::internal::ClientCallbackReaderWriterFactory< ::iv::CloudSwapRequestStream,::iv::CloudSwapReplyStream>::Create(stub_->channel_.get(), stub_->rpcmethod_swap_, context, reactor);
+}
+
+::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* CloudSwapStream::Stub::AsyncswapRaw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) {
+  return ::grpc::internal::ClientAsyncReaderWriterFactory< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>::Create(channel_.get(), cq, rpcmethod_swap_, context, true, tag);
+}
+
+::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* CloudSwapStream::Stub::PrepareAsyncswapRaw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) {
+  return ::grpc::internal::ClientAsyncReaderWriterFactory< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>::Create(channel_.get(), cq, rpcmethod_swap_, context, false, nullptr);
+}
+
+CloudSwapStream::Service::Service() {
+  AddMethod(new ::grpc::internal::RpcServiceMethod(
+      CloudSwapStream_method_names[0],
+      ::grpc::internal::RpcMethod::BIDI_STREAMING,
+      new ::grpc::internal::BidiStreamingHandler< CloudSwapStream::Service, ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>(
+          [](CloudSwapStream::Service* service,
+             ::grpc::ServerContext* ctx,
+             ::grpc::ServerReaderWriter<::iv::CloudSwapReplyStream,
+             ::iv::CloudSwapRequestStream>* stream) {
+               return service->swap(ctx, stream);
+             }, this)));
+}
+
+CloudSwapStream::Service::~Service() {
+}
+
+::grpc::Status CloudSwapStream::Service::swap(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* stream) {
+  (void) context;
+  (void) stream;
+  return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+}
+
+
+}  // namespace iv
+

src/driver/driver_cloud_swap_server/cloudswap.grpc.pb.h

@@ -0,0 +1,279 @@
+// Generated by the gRPC C++ plugin.
+// If you make any local change, they will be lost.
+// source: cloudswap.proto
+// Original file comments:
+// Copyright 2015 gRPC authors.
+//
+// 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.
+//
+#ifndef GRPC_cloudswap_2eproto__INCLUDED
+#define GRPC_cloudswap_2eproto__INCLUDED
+
+#include "cloudswap.pb.h"
+
+#include <functional>
+#include <grpc/impl/codegen/port_platform.h>
+#include <grpcpp/impl/codegen/async_generic_service.h>
+#include <grpcpp/impl/codegen/async_stream.h>
+#include <grpcpp/impl/codegen/async_unary_call.h>
+#include <grpcpp/impl/codegen/client_callback.h>
+#include <grpcpp/impl/codegen/client_context.h>
+#include <grpcpp/impl/codegen/completion_queue.h>
+#include <grpcpp/impl/codegen/message_allocator.h>
+#include <grpcpp/impl/codegen/method_handler.h>
+#include <grpcpp/impl/codegen/proto_utils.h>
+#include <grpcpp/impl/codegen/rpc_method.h>
+#include <grpcpp/impl/codegen/server_callback.h>
+#include <grpcpp/impl/codegen/server_callback_handlers.h>
+#include <grpcpp/impl/codegen/server_context.h>
+#include <grpcpp/impl/codegen/service_type.h>
+#include <grpcpp/impl/codegen/status.h>
+#include <grpcpp/impl/codegen/stub_options.h>
+#include <grpcpp/impl/codegen/sync_stream.h>
+
+namespace iv {
+
+// The Upload service definition.
+class CloudSwapStream final {
+ public:
+  static constexpr char const* service_full_name() {
+    return "iv.CloudSwapStream";
+  }
+  class StubInterface {
+   public:
+    virtual ~StubInterface() {}
+    // Sends a Upload
+    std::unique_ptr< ::grpc::ClientReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>> swap(::grpc::ClientContext* context) {
+      return std::unique_ptr< ::grpc::ClientReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>>(swapRaw(context));
+    }
+    std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>> Asyncswap(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) {
+      return std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>>(AsyncswapRaw(context, cq, tag));
+    }
+    std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>> PrepareAsyncswap(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>>(PrepareAsyncswapRaw(context, cq));
+    }
+    class experimental_async_interface {
+     public:
+      virtual ~experimental_async_interface() {}
+      // Sends a Upload
+      #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+      virtual void swap(::grpc::ClientContext* context, ::grpc::ClientBidiReactor< ::iv::CloudSwapRequestStream,::iv::CloudSwapReplyStream>* reactor) = 0;
+      #else
+      virtual void swap(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor< ::iv::CloudSwapRequestStream,::iv::CloudSwapReplyStream>* reactor) = 0;
+      #endif
+    };
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+    typedef class experimental_async_interface async_interface;
+    #endif
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+    async_interface* async() { return experimental_async(); }
+    #endif
+    virtual class experimental_async_interface* experimental_async() { return nullptr; }
+  private:
+    virtual ::grpc::ClientReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* swapRaw(::grpc::ClientContext* context) = 0;
+    virtual ::grpc::ClientAsyncReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* AsyncswapRaw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) = 0;
+    virtual ::grpc::ClientAsyncReaderWriterInterface< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* PrepareAsyncswapRaw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) = 0;
+  };
+  class Stub final : public StubInterface {
+   public:
+    Stub(const std::shared_ptr< ::grpc::ChannelInterface>& channel);
+    std::unique_ptr< ::grpc::ClientReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>> swap(::grpc::ClientContext* context) {
+      return std::unique_ptr< ::grpc::ClientReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>>(swapRaw(context));
+    }
+    std::unique_ptr<  ::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>> Asyncswap(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) {
+      return std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>>(AsyncswapRaw(context, cq, tag));
+    }
+    std::unique_ptr<  ::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>> PrepareAsyncswap(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) {
+      return std::unique_ptr< ::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>>(PrepareAsyncswapRaw(context, cq));
+    }
+    class experimental_async final :
+      public StubInterface::experimental_async_interface {
+     public:
+      #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+      void swap(::grpc::ClientContext* context, ::grpc::ClientBidiReactor< ::iv::CloudSwapRequestStream,::iv::CloudSwapReplyStream>* reactor) override;
+      #else
+      void swap(::grpc::ClientContext* context, ::grpc::experimental::ClientBidiReactor< ::iv::CloudSwapRequestStream,::iv::CloudSwapReplyStream>* reactor) override;
+      #endif
+     private:
+      friend class Stub;
+      explicit experimental_async(Stub* stub): stub_(stub) { }
+      Stub* stub() { return stub_; }
+      Stub* stub_;
+    };
+    class experimental_async_interface* experimental_async() override { return &async_stub_; }
+
+   private:
+    std::shared_ptr< ::grpc::ChannelInterface> channel_;
+    class experimental_async async_stub_{this};
+    ::grpc::ClientReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* swapRaw(::grpc::ClientContext* context) override;
+    ::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* AsyncswapRaw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq, void* tag) override;
+    ::grpc::ClientAsyncReaderWriter< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* PrepareAsyncswapRaw(::grpc::ClientContext* context, ::grpc::CompletionQueue* cq) override;
+    const ::grpc::internal::RpcMethod rpcmethod_swap_;
+  };
+  static std::unique_ptr<Stub> NewStub(const std::shared_ptr< ::grpc::ChannelInterface>& channel, const ::grpc::StubOptions& options = ::grpc::StubOptions());
+
+  class Service : public ::grpc::Service {
+   public:
+    Service();
+    virtual ~Service();
+    // Sends a Upload
+    virtual ::grpc::Status swap(::grpc::ServerContext* context, ::grpc::ServerReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* stream);
+  };
+  template <class BaseClass>
+  class WithAsyncMethod_swap : public BaseClass {
+   private:
+    void BaseClassMustBeDerivedFromService(const Service* /*service*/) {}
+   public:
+    WithAsyncMethod_swap() {
+      ::grpc::Service::MarkMethodAsync(0);
+    }
+    ~WithAsyncMethod_swap() override {
+      BaseClassMustBeDerivedFromService(this);
+    }
+    // disable synchronous version of this method
+    ::grpc::Status swap(::grpc::ServerContext* /*context*/, ::grpc::ServerReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* /*stream*/)  override {
+      abort();
+      return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+    }
+    void Requestswap(::grpc::ServerContext* context, ::grpc::ServerAsyncReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* stream, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) {
+      ::grpc::Service::RequestAsyncBidiStreaming(0, context, stream, new_call_cq, notification_cq, tag);
+    }
+  };
+  typedef WithAsyncMethod_swap<Service > AsyncService;
+  template <class BaseClass>
+  class ExperimentalWithCallbackMethod_swap : public BaseClass {
+   private:
+    void BaseClassMustBeDerivedFromService(const Service* /*service*/) {}
+   public:
+    ExperimentalWithCallbackMethod_swap() {
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+      ::grpc::Service::
+    #else
+      ::grpc::Service::experimental().
+    #endif
+        MarkMethodCallback(0,
+          new ::grpc::internal::CallbackBidiHandler< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>(
+            [this](
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+                   ::grpc::CallbackServerContext*
+    #else
+                   ::grpc::experimental::CallbackServerContext*
+    #endif
+                     context) { return this->swap(context); }));
+    }
+    ~ExperimentalWithCallbackMethod_swap() override {
+      BaseClassMustBeDerivedFromService(this);
+    }
+    // disable synchronous version of this method
+    ::grpc::Status swap(::grpc::ServerContext* /*context*/, ::grpc::ServerReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* /*stream*/)  override {
+      abort();
+      return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+    }
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+    virtual ::grpc::ServerBidiReactor< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* swap(
+      ::grpc::CallbackServerContext* /*context*/)
+    #else
+    virtual ::grpc::experimental::ServerBidiReactor< ::iv::CloudSwapRequestStream, ::iv::CloudSwapReplyStream>* swap(
+      ::grpc::experimental::CallbackServerContext* /*context*/)
+    #endif
+      { return nullptr; }
+  };
+  #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+  typedef ExperimentalWithCallbackMethod_swap<Service > CallbackService;
+  #endif
+
+  typedef ExperimentalWithCallbackMethod_swap<Service > ExperimentalCallbackService;
+  template <class BaseClass>
+  class WithGenericMethod_swap : public BaseClass {
+   private:
+    void BaseClassMustBeDerivedFromService(const Service* /*service*/) {}
+   public:
+    WithGenericMethod_swap() {
+      ::grpc::Service::MarkMethodGeneric(0);
+    }
+    ~WithGenericMethod_swap() override {
+      BaseClassMustBeDerivedFromService(this);
+    }
+    // disable synchronous version of this method
+    ::grpc::Status swap(::grpc::ServerContext* /*context*/, ::grpc::ServerReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* /*stream*/)  override {
+      abort();
+      return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+    }
+  };
+  template <class BaseClass>
+  class WithRawMethod_swap : public BaseClass {
+   private:
+    void BaseClassMustBeDerivedFromService(const Service* /*service*/) {}
+   public:
+    WithRawMethod_swap() {
+      ::grpc::Service::MarkMethodRaw(0);
+    }
+    ~WithRawMethod_swap() override {
+      BaseClassMustBeDerivedFromService(this);
+    }
+    // disable synchronous version of this method
+    ::grpc::Status swap(::grpc::ServerContext* /*context*/, ::grpc::ServerReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* /*stream*/)  override {
+      abort();
+      return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+    }
+    void Requestswap(::grpc::ServerContext* context, ::grpc::ServerAsyncReaderWriter< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* stream, ::grpc::CompletionQueue* new_call_cq, ::grpc::ServerCompletionQueue* notification_cq, void *tag) {
+      ::grpc::Service::RequestAsyncBidiStreaming(0, context, stream, new_call_cq, notification_cq, tag);
+    }
+  };
+  template <class BaseClass>
+  class ExperimentalWithRawCallbackMethod_swap : public BaseClass {
+   private:
+    void BaseClassMustBeDerivedFromService(const Service* /*service*/) {}
+   public:
+    ExperimentalWithRawCallbackMethod_swap() {
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+      ::grpc::Service::
+    #else
+      ::grpc::Service::experimental().
+    #endif
+        MarkMethodRawCallback(0,
+          new ::grpc::internal::CallbackBidiHandler< ::grpc::ByteBuffer, ::grpc::ByteBuffer>(
+            [this](
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+                   ::grpc::CallbackServerContext*
+    #else
+                   ::grpc::experimental::CallbackServerContext*
+    #endif
+                     context) { return this->swap(context); }));
+    }
+    ~ExperimentalWithRawCallbackMethod_swap() override {
+      BaseClassMustBeDerivedFromService(this);
+    }
+    // disable synchronous version of this method
+    ::grpc::Status swap(::grpc::ServerContext* /*context*/, ::grpc::ServerReaderWriter< ::iv::CloudSwapReplyStream, ::iv::CloudSwapRequestStream>* /*stream*/)  override {
+      abort();
+      return ::grpc::Status(::grpc::StatusCode::UNIMPLEMENTED, "");
+    }
+    #ifdef GRPC_CALLBACK_API_NONEXPERIMENTAL
+    virtual ::grpc::ServerBidiReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* swap(
+      ::grpc::CallbackServerContext* /*context*/)
+    #else
+    virtual ::grpc::experimental::ServerBidiReactor< ::grpc::ByteBuffer, ::grpc::ByteBuffer>* swap(
+      ::grpc::experimental::CallbackServerContext* /*context*/)
+    #endif
+      { return nullptr; }
+  };
+  typedef Service StreamedUnaryService;
+  typedef Service SplitStreamedService;
+  typedef Service StreamedService;
+};
+
+}  // namespace iv
+
+
+#endif  // GRPC_cloudswap_2eproto__INCLUDED

src/driver/driver_cloud_swap_server/driver_cloud_swap_server.pro

@@ -0,0 +1,53 @@
+QT -= gui
+
+CONFIG += c++11 console
+CONFIG -= app_bundle
+
+QMAKE_LFLAGS += -no-pie
+
+
+# The following define makes your compiler emit warnings if you use
+# any Qt feature that has been marked deprecated (the exact warnings
+# depend on your compiler). Please consult the documentation of the
+# deprecated API in order to know how to port your code away from it.
+DEFINES += QT_DEPRECATED_WARNINGS
+
+# You can also make your code fail to compile if it uses deprecated APIs.
+# In order to do so, uncomment the following line.
+# You can also select to disable deprecated APIs only up to a certain version of Qt.
+#DEFINES += QT_DISABLE_DEPRECATED_BEFORE=0x060000    # disables all the APIs deprecated before Qt 6.0.0
+
+SOURCES += \
+        main.cpp \
+    cloudswap.grpc.pb.cc \
+    swapserver.cpp \
+    swapunit.cpp \
+    ../../include/msgtype/cloudswap.pb.cc
+# Default rules for deployment.
+qnx: target.path = /tmp/$${TARGET}/bin
+else: unix:!android: target.path = /opt/$${TARGET}/bin
+!isEmpty(target.path): INSTALLS += target
+
+!include(../../../include/common.pri ) {
+    error( "Couldn't find the common.pri file!" )
+}
+
+!include(../../../include/ivprotobuf.pri ) {
+    error( "Couldn't find the ivprotobuf.pri file!" )
+}
+
+!include(../../../include/ivboost.pri ) {
+    error( "Couldn't find the ivboost.pri file!" )
+}
+
+!include(../../../include/ivgrpc.pri ) {
+    error( "Couldn't find the ivgrpc.pri file!" )
+}
+
+
+HEADERS += \
+    cloudswap.grpc.pb.h \
+    swapserver.h \
+    swapunit.h \
+    ../../include/msgtype/cloudswap.pb.h
+

src/driver/driver_cloud_swap_server/main.cpp

@@ -0,0 +1,182 @@
+#include <QCoreApplication>
+
+#include <iostream>
+#include <vector>
+
+#include <iostream>
+#include <memory>
+#include <string>
+
+#include <grpcpp/grpcpp.h>
+#include <grpcpp/health_check_service_interface.h>
+#include <grpcpp/ext/proto_server_reflection_plugin.h>
+
+#include "cloudswap.grpc.pb.h"
+
+#include "swapserver.h"
+
+using grpc::Server;
+using grpc::ServerBuilder;
+using grpc::ServerContext;
+using grpc::Status;
+
+#include <thread>
+
+char gstrport[255];
+
+swapserver * gpswapserver;
+
+#include <getopt.h>
+
+void print_useage()
+{
+    std::cout<<" -p --port $port : port . eq.  -p 50051"<<std::endl;
+    std::cout<<" -h --help print help"<<std::endl;
+}
+
+int  GetOptLong(int argc, char *argv[]) {
+    int nRtn = 0;
+    int opt; // getopt_long() 的返回值
+//    int digit_optind = 0; // 设置短参数类型及是否需要参数
+
+    // 如果option_index非空，它指向的变量将记录当前找到参数符合long_opts里的
+    // 第几个元素的描述，即是long_opts的下标值
+    int option_index = 0;
+    // 设置短参数类型及是否需要参数
+    const char *optstring = "m:r:x:y:o:p:s:h";
+
+
+    static struct option long_options[] = {
+        {"port", required_argument, NULL, 'p'},
+        {"help",  no_argument,       NULL, 'h'},
+ //       {"optarg", optional_argument, NULL, 'o'},
+        {0, 0, 0, 0}  // 添加 {0, 0, 0, 0} 是为了防止输入空值
+    };
+
+    while ( (opt = getopt_long(argc,
+                               argv,
+                               optstring,
+                               long_options,
+                               &option_index)) != -1) {
+        switch(opt)
+        {
+
+        case 'p':
+            strncpy(gstrport,optarg,255);
+            break;
+        case 'h':
+            print_useage();
+            nRtn = 1; //because use -h
+            break;
+        default:
+            break;
+        }
+
+    }
+
+    return nRtn;
+}
+
+
+// Logic and data behind the server's behavior.
+class CloudSwapServiceImpl final : public iv::CloudSwapStream::Service {
+  Status swap(ServerContext* context, ::grpc::ServerReaderWriter<iv::CloudSwapReplyStream, iv::CloudSwapRequestStream>* stream) override {
+
+      (void)context;
+      iv::CloudSwapRequestStream request;
+
+      bool bCreateThread = false;
+      std::string strnodeid;
+      std::string strobjnodeid;
+      swapunit * pswap = NULL;
+
+
+      while (stream->Read(&request))
+      {
+
+          if(bCreateThread == false)
+          {
+              strnodeid = request.strnodeid();
+              strobjnodeid = request.strobjnodeid();
+              pswap = gpswapserver->AddSwapUnit(strnodeid,strobjnodeid,stream);
+              bCreateThread = true;
+          }
+
+          pswap->SetPingAvg(request.pingavg());
+          int ndatasize = request.xdata().size();
+          if(request.nunitcount() > 0)
+          {
+              ndatasize = request.xdata().size();
+              std::shared_ptr<char> pdata_ptr = std::shared_ptr<char>(new char[ndatasize]);
+              memcpy(pdata_ptr.get(),request.xdata().data(),ndatasize);
+              gpswapserver->broadmsg(strobjnodeid,pdata_ptr,ndatasize,request.pingavg());
+          }
+          else
+          {
+              std::cout<<"receive heartbeat from "<<strnodeid<<std::endl;
+          }
+
+
+      }
+      pswap->stopswap();  //Because connection fail.
+      std::cout<<" no conn"<<std::endl;
+      std::cout<<"dis connect."<<std::endl;
+
+    return Status::OK;
+  }
+
+
+
+};
+
+void RunServer() {
+  std::string server_address("0.0.0.0:");
+  server_address = server_address.append(gstrport);
+  CloudSwapServiceImpl service;
+
+  grpc::EnableDefaultHealthCheckService(true);
+//  grpc::reflection::InitProtoReflectionServerBuilderPlugin();
+  ServerBuilder builder;
+  // Listen on the given address without any authentication mechanism.
+  builder.AddListeningPort(server_address, grpc::InsecureServerCredentials());
+
+  builder.SetMaxReceiveMessageSize(300000000);
+ // builder.SetMaxMessageSize(100000000);
+
+
+//  builder.SetMaxSendMessageSize(100000000);
+  // Register "service" as the instance through which we'll communicate with
+  // clients. In this case it corresponds to an *synchronous* service.
+  builder.RegisterService(&service);
+  // Finally assemble the server.
+  std::unique_ptr<Server> server(builder.BuildAndStart());
+  std::cout << "Server listening on " << server_address << std::endl;
+
+  // Wait for the server to shutdown. Note that some other thread must be
+  // responsible for shutting down the server for this call to ever return.
+  server->Wait();
+}
+
+
+
+
+int main(int argc, char *argv[])
+{
+    QCoreApplication a(argc, argv);
+
+    snprintf(gstrport,255,"50061");
+
+    int nRtn = GetOptLong(argc,argv);
+    if(nRtn == 1)  //show help,so exit.
+    {
+        return 0;
+    }
+
+
+    gpswapserver = new swapserver();
+
+    RunServer();
+
+
+    return a.exec();
+}

src/driver/driver_cloud_swap_server/swapserver.cpp

@@ -0,0 +1,26 @@
+#include "swapserver.h"
+
+swapserver::swapserver()
+{
+
+}
+
+int swapserver::broadmsg(std::string strobjid,std::shared_ptr<char> pdata_ptr,int ndatasize,double fpingavg)
+{
+    int nrtn = 0;
+    int i;
+    for(i=0;i<(int)mvectorswap.size();i++)
+    {
+        nrtn = nrtn + mvectorswap[i]->sendmsg(strobjid,pdata_ptr,ndatasize,fpingavg);
+    }
+    return nrtn;
+}
+
+swapunit * swapserver::AddSwapUnit(std::string strnodeid, std::string strobjnodeid, ::grpc::ServerReaderWriter<iv::CloudSwapReplyStream, iv::CloudSwapRequestStream> *stream)
+{
+    swapunit * pswap = new swapunit(strnodeid,strobjnodeid,stream);
+    mmutex.lock();
+    mvectorswap.push_back(pswap);
+    mmutex.unlock();
+    return pswap;
+}

src/driver/driver_cloud_swap_server/swapserver.h

@@ -0,0 +1,22 @@
+#ifndef SWAPSERVER_H
+#define SWAPSERVER_H
+
+#include <memory>
+#include <swapunit.h>
+#include <mutex>
+
+class swapserver
+{
+public:
+    swapserver();
+
+private:
+    std::vector<swapunit *> mvectorswap;
+    std::mutex mmutex;
+
+public:
+    int broadmsg(std::string strobjid,std::shared_ptr<char> pdata_ptr,int ndatasize,double fpingavg);
+    swapunit * AddSwapUnit(std::string strnodeid,std::string strobjnodeid,::grpc::ServerReaderWriter<iv::CloudSwapReplyStream, iv::CloudSwapRequestStream>* stream);
+};
+
+#endif // SWAPSERVER_H

src/driver/driver_cloud_swap_server/swapunit.cpp

@@ -0,0 +1,118 @@
+#include "swapunit.h"
+
+swapunit::swapunit(std::string strnodeid,std::string strobjnodeid,::grpc::ServerReaderWriter<iv::CloudSwapReplyStream, iv::CloudSwapRequestStream>* stream)
+{
+    mstrnodeid = strnodeid;
+    mstrobjnodeid = strobjnodeid;
+    mpthread = new std::thread(&swapunit::threadsend,this,stream);
+}
+
+swapunit::~swapunit()
+{
+
+}
+
+int swapunit::sendmsg(std::string strobjid, std::shared_ptr<char> pdata_ptr, int ndatasize,double fpingavg)
+{
+    if(mbrun == false)
+    {
+        return 0;
+    }
+    if(strobjid != mstrnodeid)
+    {
+        return 0;
+    }
+    iv::swapmsg xmsg;
+    xmsg.ndatasize = ndatasize;
+    xmsg.nRecvTime = std::chrono::system_clock::now().time_since_epoch().count();
+    xmsg.pdata_ptr = pdata_ptr;
+    xmsg.pingavg = fpingavg;
+
+
+    if(mvectorsendmsgbuf.size()>30000)
+    {
+        std::cout<<"may connection down, buffer now full."<<std::endl;
+        mmutex.lock();
+        mvectorsendmsgbuf.clear();
+        mmutex.unlock();
+    }
+    mmutex.lock();
+    mvectorsendmsgbuf.push_back(xmsg);
+    mmutex.unlock();
+
+    mcv.notify_all(); //notify
+    return 0;
+}
+
+void swapunit::threadsend(::grpc::ServerReaderWriter<iv::CloudSwapReplyStream, iv::CloudSwapRequestStream> *stream)
+{
+
+    int i;
+    int64_t xLastSend = std::chrono::system_clock::now().time_since_epoch().count();
+    while(mbrun)
+    {
+        std::vector<iv::swapmsg> xvectorsendmsgbuf;
+        mmutex.lock();
+        int nsize = mvectorsendmsgbuf.size();
+        for(i=0;i<nsize;i++)
+        {
+            xvectorsendmsgbuf.push_back(mvectorsendmsgbuf[i]);
+        }
+        mvectorsendmsgbuf.clear();
+        mmutex.unlock();
+        int64_t xNow = std::chrono::system_clock::now().time_since_epoch().count();
+        if((abs((xNow - xLastSend)/1000000) <1000 ) &&(xvectorsendmsgbuf.size() == 0))
+        {
+            std::unique_lock<std::mutex> lk(mmutexcv);
+            if(mcv.wait_for(lk, std::chrono::milliseconds(100)) == std::cv_status::timeout)
+            {
+                lk.unlock();
+                continue;
+            }
+            else
+            {
+                lk.unlock();
+            }
+        }
+        iv::CloudSwapReplyStream xreply;
+        xreply.set_nres(mvectorsendmsgbuf.size());
+        if(xvectorsendmsgbuf.size() == 0)
+        {
+            xreply.set_flatency_inserver(0.0);
+
+        }
+        else
+        {
+            int64_t nlatinserver = (xNow - xvectorsendmsgbuf[0].nRecvTime);
+            double flatinserver = nlatinserver;
+            flatinserver = flatinserver/1000000000.0;
+            xreply.set_flatency_inserver(flatinserver);
+        }
+        for(i=0;i<nsize;i++)
+        {
+            xreply.add_xdata(xvectorsendmsgbuf[i].pdata_ptr.get(),xvectorsendmsgbuf[i].ndatasize);
+        }
+        xreply.set_flatency_innode(0);
+        xreply.set_flatency_inobjnode(0);
+        xreply.set_nmsgservertime(xNow);
+        stream->Write(xreply);
+        xLastSend = xNow;
+
+    }
+    mbdel = true;
+}
+
+bool swapunit::CanDel()
+{
+    return mbdel;
+}
+
+void swapunit::stopswap()
+{
+    mbrun = false;
+}
+
+void swapunit::SetPingAvg(double fpingavg)
+{
+    mfpingavg = fpingavg;
+}

src/driver/driver_cloud_swap_server/swapunit.h

@@ -0,0 +1,57 @@
+#ifndef SWAPUNIT_H
+#define SWAPUNIT_H
+
+#include <memory>
+#include <string>
+#include <vector>
+#include <mutex>
+#include <thread>
+#include <condition_variable>
+
+#include "cloudswap.grpc.pb.h"
+
+
+namespace  iv {
+
+struct swapmsg
+{
+    std::shared_ptr<char> pdata_ptr;
+    int ndatasize;
+    int64_t nRecvTime;
+    double pingavg ;
+};
+
+}
+
+class swapunit
+{
+public:
+    swapunit(std::string strnodeid,std::string strobjnodeid,::grpc::ServerReaderWriter<iv::CloudSwapReplyStream, iv::CloudSwapRequestStream>* stream);
+    ~swapunit();
+
+private:
+    void threadsend(::grpc::ServerReaderWriter<iv::CloudSwapReplyStream, iv::CloudSwapRequestStream>* stream);
+
+private:
+    std::string mstrnodeid;
+    std::string mstrobjnodeid;
+    bool mbrun = true;
+    bool mbdel = false;
+    std::thread * mpthread;
+    double mfpingavg = 0; //ms
+
+private:
+    std::vector<iv::swapmsg> mvectorsendmsgbuf;
+    std::mutex mmutex;
+
+    std::mutex mmutexcv;
+    std::condition_variable mcv;
+
+public:
+    int sendmsg(std::string strobjid,std::shared_ptr<char> pdata_ptr,int ndatasize,double fpingavg);
+    bool CanDel();
+    void stopswap();
+    void SetPingAvg(double fpingavg);
+};
+
+#endif // SWAPUNIT_H