diff --git a/CMakeLists.txt b/CMakeLists.txt
index ef0b9a0..75eab27 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -139,6 +139,8 @@ install(TARGETS ${BIN_NAME} DESTINATION bin)
 set(BIN_NAME dde-clipboard-daemon)
 
 file(GLOB_RECURSE dde-clipboard-daemon_SCRS
+   "dde-clipboard-daemon/serviceflow/*.h"
+   "dde-clipboard-daemon/serviceflow/*.cpp"
    "dde-clipboard-daemon/*.h"
    "dde-clipboard-daemon/*.cpp"
 )
diff --git a/dde-clipboard-daemon/clipboardloader.cpp b/dde-clipboard-daemon/clipboardloader.cpp
index fadb925..2fdffaf 100644
--- a/dde-clipboard-daemon/clipboardloader.cpp
+++ b/dde-clipboard-daemon/clipboardloader.cpp
@@ -90,7 +90,6 @@ ClipboardLoader::ClipboardLoader(QObject *parent)
     if (qEnvironmentVariable("XDG_SESSION_TYPE").contains("wayland")) {
 #ifdef USE_DEEPIN_KF5_WAYLAND
         m_waylandCopyClient = new WaylandCopyClient(this);
-        m_waylandCopyClient->init();
 
         connect(m_waylandCopyClient, &WaylandCopyClient::dataChanged, this, [this] {
             this->doWork(WAYLAND_PROTOCOL);
diff --git a/dde-clipboard-daemon/serviceflow/command.cpp b/dde-clipboard-daemon/serviceflow/command.cpp
new file mode 100644
index 0000000..32aaf43
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/command.cpp
@@ -0,0 +1,73 @@
+#include "command.h"
+#include "servicetool.h"
+#include <QThread>
+
+Command::Command(CommandService *service)
+    : m_service(service)
+{
+    setAutoDelete(true);
+}
+
+void Command::exit()
+{
+    m_isExit = true;
+}
+
+Command::State Command::state() const
+{
+    return m_state;
+}
+
+CommandService *Command::service() const
+{
+    return m_service;
+}
+
+void Command::installMessageExtractor(Extractor *extractor)
+{
+    Q_ASSERT(extractor);
+    switch (extractor->direction()) {
+    case Extractor::CommandToTransporter: {
+        if (m_outEx) {
+            qWarning("This command already has the out direction Extractor.");
+            return;
+        }
+        m_outEx = extractor;
+    }
+        break;
+    case Extractor::TransporterToCommand: {
+        if (m_inEx) {
+            qWarning("This command already has the in direction Extractor.");
+            return;
+        }
+        m_inEx = extractor;
+    }
+        break;
+    }
+}
+
+void Command::run()
+{
+    Q_ASSERT(m_inEx);
+    while (true) {
+        m_state = Preparing;
+        CommandMessage *msg = m_inEx->takeFromTransporter();
+        if (m_isExit) {
+            delete msg;
+            break;
+        }
+
+        m_state = Running;
+        QList<CommandMessage *> destMsgs = doExecute(msg);
+
+        // 允许复用同一个 msg 的情况
+        if (!destMsgs.contains(msg))
+            delete msg;
+
+        if (m_outEx) {
+            for (auto destMsg : destMsgs)
+                m_outEx->pushToTransporter(destMsg);
+        }
+    }
+    m_state = Preparing;
+}
diff --git a/dde-clipboard-daemon/serviceflow/command.h b/dde-clipboard-daemon/serviceflow/command.h
new file mode 100644
index 0000000..dad43c9
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/command.h
@@ -0,0 +1,37 @@
+#ifndef COMMAND
+#define COMMAND
+
+#include <QRunnable>
+
+class CommandService;
+class CommandMessage;
+class Extractor;
+class Command : public QRunnable
+{
+public:
+    enum State {
+        Preparing,
+        Running
+    };
+    Command(CommandService *service = nullptr);
+
+    virtual QList<CommandMessage *> doExecute(CommandMessage *) = 0;
+    virtual QString name() const = 0;
+
+    void exit();
+    State state() const;
+
+    CommandService *service() const;
+    void installMessageExtractor(Extractor *extractor);
+
+private:
+    void run() override;
+
+    Extractor * m_inEx = nullptr;
+    Extractor * m_outEx = nullptr;
+    bool m_isExit = false;
+    State m_state = Preparing;
+    CommandService *m_service = nullptr;
+};
+
+#endif  // COMMANDMESSAGE
diff --git a/dde-clipboard-daemon/serviceflow/commandmessage.h b/dde-clipboard-daemon/serviceflow/commandmessage.h
new file mode 100644
index 0000000..b918a51
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/commandmessage.h
@@ -0,0 +1,27 @@
+#ifndef COMMANDMESSAGE
+#define COMMANDMESSAGE
+
+class CommandMessage
+{
+public:
+    enum Error {
+        NoError,
+        ExecuteError
+    };
+
+    CommandMessage() = default;
+    virtual ~CommandMessage() {}
+
+    inline Error error() const {
+        return m_error;
+    }
+
+    inline void setError(Error err) {
+        m_error = err;
+    }
+
+protected:
+    Error m_error = NoError;
+};
+
+#endif  // COMMANDMESSAGE
diff --git a/dde-clipboard-daemon/serviceflow/commandservice.cpp b/dde-clipboard-daemon/serviceflow/commandservice.cpp
new file mode 100644
index 0000000..a5c7a00
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/commandservice.cpp
@@ -0,0 +1,21 @@
+#include "commandservice.h"
+
+CommandService::CommandService()
+{
+
+}
+
+CommandService::~CommandService()
+{
+
+}
+
+CommandService *CommandService::nextService() const
+{
+    return this->m_nextService;
+}
+
+void CommandService::setNextService(CommandService *service)
+{
+    this->m_nextService = service;
+}
diff --git a/dde-clipboard-daemon/serviceflow/commandservice.h b/dde-clipboard-daemon/serviceflow/commandservice.h
new file mode 100644
index 0000000..1f991e1
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/commandservice.h
@@ -0,0 +1,21 @@
+#ifndef COMMANDSERVICE
+#define COMMANDSERVICE
+
+#include "command.h"
+
+class CommandService
+{
+public:
+    CommandService();
+    virtual ~CommandService();
+
+    virtual QList<Command *> commands() = 0;
+
+    CommandService *nextService() const;
+    void setNextService(CommandService* service);
+
+private:
+    CommandService* m_nextService;
+};
+
+#endif  // COMMANDSERVICE
diff --git a/dde-clipboard-daemon/serviceflow/servicemanager.cpp b/dde-clipboard-daemon/serviceflow/servicemanager.cpp
new file mode 100644
index 0000000..d4e67d6
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/servicemanager.cpp
@@ -0,0 +1,297 @@
+#include "servicemanager.h"
+#include "servicetool.h"
+
+#include <QThreadPool>
+
+// ### Hidden for using
+class ServiceFlow
+{
+public:
+    ServiceFlow(CommandService *service);
+    ~ServiceFlow();
+
+    QList<CommandService *> services() const;
+    void start();
+    void appendFlowMessage(CommandMessage *msg);
+    bool isStarted() const;
+
+private:
+    void initServiceCommands();
+
+    void setupFlow();
+    void setupHeaderFlow();
+
+private:
+    CommandService *m_root;
+    Transporter *m_rootTrans;
+    QList<Transporter *> m_transporters;
+    QList<Extractor *> m_extractors;
+    QMap<CommandService *, QList<Command *>> m_serviceCommands;
+    QThreadPool *m_threadPool;
+};
+
+ServiceFlow::ServiceFlow(CommandService *service)
+    : m_root(service)
+    , m_rootTrans(new Transporter)
+    , m_threadPool(new QThreadPool)
+{
+    initServiceCommands();
+    setupFlow();
+}
+
+ServiceFlow::~ServiceFlow()
+{
+    for (auto machines : m_serviceCommands) {
+        for (auto machine : qAsConst(machines)) {
+            machine->exit();
+        }
+    }
+
+    m_rootTrans->release();
+    for (auto trans : m_transporters)
+        trans->release();
+
+    m_threadPool->waitForDone();
+
+    delete m_rootTrans;
+    qDeleteAll(m_transporters);
+    qDeleteAll(m_extractors);
+    delete m_threadPool;
+
+}
+
+QList<CommandService *> ServiceFlow::services() const
+{
+    return m_serviceCommands.keys();
+}
+
+void ServiceFlow::start()
+{
+    QList<CommandService *> stores = { m_root };
+    CommandService *header = nullptr;
+    int priority = 0;
+    while (!stores.isEmpty()) {
+        int n = stores.size();
+        for (int i = 0; i < n; ++i) {
+            header = stores.takeFirst();
+            for (auto machine : m_serviceCommands.value(header)) {
+                if (machine->state() != Command::Preparing)
+                    continue;
+                m_threadPool->start(machine, priority);
+            }
+            if (auto nextService = header->nextService())
+                stores.append(nextService);
+        }
+        priority++;
+    }
+
+}
+
+void ServiceFlow::appendFlowMessage(CommandMessage *msg)
+{
+    m_rootTrans->putMessage(msg);
+}
+
+bool ServiceFlow::isStarted() const
+{
+    return m_threadPool->activeThreadCount() > 0;
+}
+
+void ServiceFlow::initServiceCommands()
+{
+    m_serviceCommands.clear();
+    QList<CommandService *> headers = { m_root };
+    CommandService *header = nullptr;
+    while (!headers.isEmpty()) {
+        header = headers.takeFirst();
+        do {
+            if (m_serviceCommands.contains(header))
+                break;
+            m_serviceCommands.insert(header, header->commands());
+        } while (false);
+
+        if (auto nextService = header->nextService())
+            headers.append(nextService);
+    }
+
+    // 根据服务中命令的个数，动态创建线程池的数量
+    int accu = 0;
+    for (auto ms : m_serviceCommands.values())
+        accu += ms.count();
+    m_threadPool->setMaxThreadCount(accu);
+}
+
+void ServiceFlow::setupFlow()
+{
+    setupHeaderFlow();
+    QList<CommandService*> headers = { m_root };
+    CommandService *header = nullptr;
+
+    while (!headers.isEmpty()) {
+        header = headers.takeFirst();
+        auto next = header->nextService();
+        if (!next)
+            continue;
+        Transporter *inTrans = new Transporter();
+        m_transporters.append(inTrans);
+
+        Extractor *exOnIn = new Extractor(inTrans);
+        exOnIn->setDirection(Extractor::CommandToTransporter);
+        m_extractors.append(exOnIn);
+
+        for (auto machine : m_serviceCommands.value(header))
+            machine->installMessageExtractor(exOnIn);
+
+        Transporter *trans = inTrans;
+        Extractor *exOnOut = new Extractor(trans);
+        exOnOut->setDirection(Extractor::TransporterToCommand);
+        m_extractors.append(exOnOut);
+
+        for (auto machine : m_serviceCommands.value(next))
+            machine->installMessageExtractor(exOnOut);
+
+        headers.append(next);
+    }
+}
+
+void ServiceFlow::setupHeaderFlow()
+{
+    Extractor *rootExtra = new Extractor(m_rootTrans, Extractor::TransporterToCommand);
+    m_extractors.append(rootExtra);
+    for (auto machine : m_serviceCommands.value(m_root))
+        machine->installMessageExtractor(rootExtra);
+}
+
+
+CommandServiceManager::CommandServiceManager()
+    : m_maxServiceCount(0)
+{
+
+}
+
+CommandServiceManager::~CommandServiceManager()
+{
+    for (auto service: m_rootServices.keys())
+        unregisterService(service);
+}
+
+void CommandServiceManager::registerService(CommandService *root)
+{
+    Q_ASSERT(root);
+    if (isServiceRegistered(root)) {
+        qWarning("Already register this service.");
+        return;
+    }
+
+    ServiceFlow *service = new ServiceFlow(root);
+    m_rootServices.insert(root, service);
+}
+
+void CommandServiceManager::unregisterService(CommandService *root)
+{
+    if (!m_rootServices.contains(root))
+        return;
+    auto const &flow = m_rootServices.take(root);
+    delete flow;
+
+    if (m_waitingServices.isEmpty())
+        return;
+
+    auto service = m_waitingServices.takeFirst();
+    start(service);
+}
+
+bool CommandServiceManager::isServiceRegistered(CommandService *service)
+{
+    if (m_rootServices.contains(service))
+        return true;
+
+    for (auto root : m_rootServices) {
+        auto const &services = root->services();
+        bool contains = std::any_of(services.cbegin(), services.cend(),
+                                    [=](const CommandService *s) {
+            return service == s;
+        });
+
+        if (!contains)
+            continue;
+        return contains;
+    }
+    return false;
+}
+
+int CommandServiceManager::registeredServicesCount() const
+{
+    return m_rootServices.count();
+}
+
+int CommandServiceManager::activeServiceFlowsCount() const
+{
+    return std::accumulate(m_rootServices.cbegin(), m_rootServices.cend(), 0,
+                           [](int prev, ServiceFlow *flow) {
+        if (flow->isStarted()) {
+            prev++;
+        }
+
+        return prev;
+    });
+}
+
+void CommandServiceManager::setMaxServiceCount(int maxCount)
+{
+    m_maxServiceCount = maxCount;
+}
+
+int CommandServiceManager::maxServiceCount() const
+{
+    return m_maxServiceCount;
+}
+
+void CommandServiceManager::start(CommandService *service)
+{
+    if (activeServiceFlowsCount() > m_maxServiceCount && m_maxServiceCount != 0) {
+        if (!m_waitingServices.contains(service))
+            m_waitingServices.append(service);
+        return;
+    }
+
+    if (CommandService *root = findRootService(service)) {
+        ServiceFlow *flow = m_rootServices.value(root);
+        if (!flow->isStarted())
+            flow->start();
+    }
+}
+
+void CommandServiceManager::appendFlowMessage(CommandMessage *message, CommandService *service)
+{
+    // 只能从 header 中输入消息流
+    Q_ASSERT(message && service);
+
+    if (!isServiceRegistered(service)) {
+        qWarning("Regiseter this service first.");
+        return;
+    }
+
+    CommandService *root = findRootService(service);
+    if (!root)
+        return;
+
+    ServiceFlow *flow = m_rootServices.value(root);
+    flow->appendFlowMessage(message);
+}
+
+CommandService *CommandServiceManager::findRootService(CommandService *service)
+{
+    for (auto root : m_rootServices) {
+        auto const &services = root->services();
+        auto const &it = std::find_if(services.begin(), services.end(),
+                                      [=](const CommandService *s) {
+            return service == s;
+        });
+
+        if (it != services.end())
+            return m_rootServices.key(root);
+    }
+
+    return nullptr;
+}
diff --git a/dde-clipboard-daemon/serviceflow/servicemanager.h b/dde-clipboard-daemon/serviceflow/servicemanager.h
new file mode 100644
index 0000000..7ea5cec
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/servicemanager.h
@@ -0,0 +1,37 @@
+#ifndef COMMANDSERVICEMANAGER
+#define COMMANDSERVICEMANAGER
+
+#include "commandservice.h"
+
+#include <QMap>
+
+class ServiceFlow;
+class CommandServiceManager
+{
+    // ### make singleton
+public:
+    CommandServiceManager();
+    ~CommandServiceManager();
+
+    void registerService(CommandService *root);
+
+    void unregisterService(CommandService *root);
+    bool isServiceRegistered(CommandService *service);
+
+    int registeredServicesCount() const;
+    int activeServiceFlowsCount() const;
+
+    void setMaxServiceCount(int maxCount);
+    int maxServiceCount() const;
+
+    void start(CommandService *service);
+    void appendFlowMessage(CommandMessage *message, CommandService *service);
+    CommandService *findRootService(CommandService *service);
+
+private:
+    QMap<CommandService *, ServiceFlow *> m_rootServices;
+    int m_maxServiceCount;
+    QList<CommandService *> m_waitingServices;
+};
+
+#endif  // COMMANDSERVICEMANAGER
diff --git a/dde-clipboard-daemon/serviceflow/servicetool.cpp b/dde-clipboard-daemon/serviceflow/servicetool.cpp
new file mode 100644
index 0000000..25a7836
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/servicetool.cpp
@@ -0,0 +1,92 @@
+#include "servicetool.h"
+
+Transporter::Transporter()
+    : m_shouldDelete(false)
+{
+
+}
+
+Transporter::~Transporter()
+{
+}
+
+void Transporter::putMessage(CommandMessage *msg)
+{
+    m_mutex.lock();
+    m_messages.append(msg);
+    m_condition.wakeAll();
+    m_mutex.unlock();
+}
+
+CommandMessage *Transporter::takeMessage()
+{
+    QMutexLocker locker(&m_mutex);
+    while (m_messages.isEmpty() && !m_shouldDelete)
+        m_condition.wait(&m_mutex);
+
+    CommandMessage *msg = nullptr;
+    if (m_messages.length() > 0)
+        msg = m_messages.takeFirst();
+
+    return msg;
+}
+
+CommandMessage *Transporter::firstMessage() const
+{
+    return messageCount() > 0 ? m_messages.first() : nullptr;
+}
+
+int Transporter::messageCount() const
+{
+    return m_messages.count();
+}
+
+void Transporter::release()
+{
+    m_mutex.lock();
+    m_shouldDelete = true;
+    m_condition.wakeAll();
+    m_mutex.unlock();
+}
+
+Extractor::Extractor(Transporter *trans)
+    : m_transporter(trans)
+{
+
+}
+
+Extractor::Extractor(Transporter *trans, Direction dir)
+    : m_direction(dir)
+    , m_transporter(trans)
+{
+
+}
+
+void Extractor::setDirection(Direction dir)
+{
+    m_direction = dir;
+}
+
+Extractor::Direction Extractor::direction() const
+{
+    return m_direction;
+}
+
+CommandMessage *Extractor::takeFromTransporter()
+{
+    if (m_direction != TransporterToCommand)
+        return nullptr;
+    return m_transporter->takeMessage();
+}
+
+void Extractor::pushToTransporter(CommandMessage *msg)
+{
+    if (m_direction != CommandToTransporter)
+        return;
+    return m_transporter->putMessage(msg);
+}
+
+void Extractor::appendCommand(Command *command)
+{
+    m_commands.append(command);
+}
diff --git a/dde-clipboard-daemon/serviceflow/servicetool.h b/dde-clipboard-daemon/serviceflow/servicetool.h
new file mode 100644
index 0000000..9e2c9cc
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/servicetool.h
@@ -0,0 +1,59 @@
+#ifndef SERVICETOOL
+#define SERVICETOOL
+
+#include "commandmessage.h"
+
+#include <QList>
+#include <QWaitCondition>
+#include <QMutex>
+
+class Transporter
+{
+public:
+    Transporter();
+    ~Transporter();
+
+    void putMessage(CommandMessage *msg);
+    CommandMessage *takeMessage();
+    CommandMessage *firstMessage() const;
+
+    int messageCount() const;
+    void release();
+
+private:
+    QList<CommandMessage *> m_messages;
+    QWaitCondition m_condition;
+    QMutex m_deleteMutex;
+    QMutex m_mutex;
+    bool m_shouldDelete;
+};
+
+class Command;
+class Extractor
+{
+    friend class Command;
+public:
+    enum Direction {
+        TransporterToCommand,
+        CommandToTransporter
+    };
+
+    Extractor(Transporter *trans);
+    Extractor(Transporter *trans, Direction dir);
+
+    void setDirection(Direction dir);
+    Direction direction() const;
+
+protected:
+    CommandMessage *takeFromTransporter();
+    void pushToTransporter(CommandMessage *msg);
+
+    void appendCommand(Command *command);
+
+private:
+    Direction m_direction;
+    QList<Command *> m_commands;
+    Transporter *m_transporter;
+};
+
+#endif  // SERVICETOOL
diff --git a/dde-clipboard-daemon/serviceflow/waylandcopyservice.cpp b/dde-clipboard-daemon/serviceflow/waylandcopyservice.cpp
new file mode 100644
index 0000000..6761bb2
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/waylandcopyservice.cpp
@@ -0,0 +1,283 @@
+#include "waylandcopyservice.h"
+#include "../waylandcopyclient.h"
+
+#include <KWayland/Client/datacontroloffer.h>
+#include <unistd.h>
+#include <QDebug>
+#include <QFile>
+#include <QImageWriter>
+#include <QImage>
+#include <QBuffer>
+#include <QUrl>
+
+QList<CommandMessage *> MimeDataFilterCommand::doExecute(CommandMessage *msg)
+{
+    auto pipeMsg = static_cast<PipeCommandMessage *>(msg);
+    const QStringList &types = filterMimeType(pipeMsg->srcMimeTypes);
+
+    if (types.isEmpty()) {
+        CommandMessage *msg = new CommandMessage();
+        msg->setError(CommandMessage::ExecuteError);
+        return {msg};
+    }
+
+    pipeMsg->dstMimeTypes = types;
+    return {pipeMsg};
+}
+
+QString MimeDataFilterCommand::name() const
+{
+    return QLatin1String("MimeDataFilterCommand");
+}
+
+QStringList MimeDataFilterCommand::filterMimeType(const QStringList &mimeTypeList)
+{
+    QStringList tmpList;
+    for (const QString &mimeType : mimeTypeList) {
+        // 根据窗管的要求，不读取纯大写、和不含'/'的字段，因为源窗口可能没有写入这些字段的数据，导致获取数据的线程一直等待。
+        if ((mimeType.contains("/") && mimeType.toUpper() != mimeType)
+                || mimeType == "FROM_DEEPIN_CLIPBOARD_MANAGER"
+                || mimeType == "TIMESTAMP") {
+            tmpList.append(mimeType);
+        }
+    }
+
+    return tmpList;
+}
+
+QList<CommandMessage *> RequestReceiveCommand::doExecute(CommandMessage *msg)
+{
+    if (msg->error() != CommandMessage::NoError)
+        return {msg};
+
+    auto pipeMsg = static_cast<PipeCommandMessage *>(msg);
+    QList<CommandMessage *> destCommandMsgs;
+
+    for (auto mimeType : pipeMsg->dstMimeTypes) {
+        int pipeFds[2];
+        if (pipe(pipeFds) != 0) {
+            qWarning() << "Create pipe failed.";
+            continue;
+        }
+
+        // 根据mime类取数据，写入pipe中
+        if (!pipeMsg->offer || !pipeMsg->offer->isValid()) {
+            close(pipeFds[0]);
+            close(pipeFds[1]);
+            pipeMsg->setError(CommandMessage::ExecuteError);
+            return {msg};
+        }
+
+        // ### lock or not ?
+        pipeMsg->offer->receive(mimeType, pipeFds[1]);
+        close(pipeFds[1]);
+
+        CopyCommandMessage *copyMsg = new CopyCommandMessage();
+        copyMsg->pipeFds[0] = pipeFds[0];
+        copyMsg->pipeFds[1] = pipeFds[1];
+        copyMsg->mimeData = pipeMsg->mimeData;
+        copyMsg->srcMimeTypes = pipeMsg->srcMimeTypes;
+        copyMsg->dstMimeTypes = pipeMsg->dstMimeTypes;
+        copyMsg->mimeTypeCount = pipeMsg->dstMimeTypes.count();
+        copyMsg->currentMimeType = mimeType;
+
+        destCommandMsgs.append(copyMsg);
+    }
+
+    if (auto recvService = dynamic_cast<RequestReceiveService *>(service()))
+        static_cast<WaylandCopyClient *>(recvService->parent())->wakePipeSyncCondition();
+
+    return destCommandMsgs;
+}
+
+QString RequestReceiveCommand::name() const
+{
+    return  QLatin1String("RequestReceiveCommand");
+}
+
+QList<CommandMessage *> ReadDataCommand::doExecute(CommandMessage *msg)
+{
+    if (msg->error() != CommandMessage::NoError)
+        return {msg};
+
+    auto copyMsg = static_cast<CopyCommandMessage *>(msg);
+    QFile readPipeDevice;
+    QByteArray data;
+
+    do {
+        if (!readPipeDevice.open(copyMsg->pipeFds[0], QIODevice::ReadOnly)) {
+            qInfo() << "Open pipe failed!";
+            break;
+        }
+
+        if (!readPipeDevice.isReadable()) {
+            qInfo() << "Pipe is not readable";
+            break;
+        }
+
+        data = readPipeDevice.readAll();
+
+        if (data.isEmpty()) {
+            qWarning() << "Pipe data is empty, mime type: " << copyMsg->currentMimeType;
+        }
+    } while (false);
+
+    close(copyMsg->pipeFds[0]);
+    copyMsg->data = data;
+    return {copyMsg};
+}
+
+QString ReadDataCommand::name() const
+{
+    return QLatin1String("ReadDataCommand");
+}
+
+QList<Command *> SyncMimeDataService::commands()
+{
+    return {new SyncMimeDataCommand(this)};
+}
+
+QList<CommandMessage *> SyncMimeDataCommand::doExecute(CommandMessage *msg)
+{
+    bool finished = false;
+    bool hasError = false;
+
+    do {
+        if (msg->error() != CommandMessage::NoError) {
+            hasError = true;
+            break;
+        }
+
+        auto copyMsg = static_cast<CopyCommandMessage *>(msg);
+        QMutexLocker locker(&m_mutex);
+        if (m_mimeCount.load() == 0) {
+            copyMsg->mimeData->clear();
+        }
+
+        copyMsg->mimeData->setData(copyMsg->currentMimeType, copyMsg->data);
+        m_mimeCount++;
+
+        if (m_mimeCount.load() == copyMsg->mimeTypeCount) {
+            // service Finished.
+            finished = true;
+        }
+    } while (false);
+
+    auto syncService = dynamic_cast<SyncMimeDataService *>(service());
+    if (finished)
+        Q_EMIT syncService->finished(!hasError);
+
+    return {};
+}
+
+QString SyncMimeDataCommand::name() const
+{
+    return QLatin1String("SyncMimeDataCommand");
+}
+
+QList<Command *> MimeDataFilterService::commands()
+{
+    return {new MimeDataFilterCommand(this)};
+}
+
+QList<Command *> RequestReceiveService::commands()
+{
+    return {new RequestReceiveCommand(this)};
+}
+
+QList<Command *> ReadDataService::commands()
+{
+    QList<Command *> commands;
+    for (int i = 0; i < MaxReadCommandCount; ++i)
+        commands.append(new ReadDataCommand(this));
+
+    return commands;
+}
+
+static QByteArray getByteArray(QMimeData *mimeData, const QString &mimeType)
+{
+    QByteArray content;
+    if (mimeType == QLatin1String("text/plain")) {
+        content = mimeData->text().toUtf8();
+    } else if (mimeData->hasImage()
+               && (mimeType == QLatin1String("application/x-qt-image")
+                   || mimeType.startsWith(QLatin1String("image/")))) {
+        QImage image = qvariant_cast<QImage>(mimeData->imageData());
+        if (!image.isNull()) {
+            QBuffer buf;
+            buf.open(QIODevice::ReadWrite);
+            QByteArray fmt = "BMP";
+            if (mimeType.startsWith(QLatin1String("image/"))) {
+                QByteArray imgFmt = mimeType.mid(6).toUpper().toLatin1();
+                if (QImageWriter::supportedImageFormats().contains(imgFmt))
+                    fmt = imgFmt;
+            }
+            QImageWriter wr(&buf, fmt);
+            wr.write(image);
+            content = buf.buffer();
+        }
+    } else if (mimeType == QLatin1String("application/x-color")) {
+        content = qvariant_cast<QColor>(mimeData->colorData()).name().toLatin1();
+    } else if (mimeType == QLatin1String("text/uri-list")) {
+        QList<QUrl> urls = mimeData->urls();
+        for (int i = 0; i < urls.count(); ++i) {
+            content.append(urls.at(i).toEncoded());
+            content.append('\n');
+        }
+    } else {
+        content = mimeData->data(mimeType);
+    }
+    return content;
+}
+
+QList<CommandMessage *> CollectWrittenDataCommand::doExecute(CommandMessage *msg)
+{
+    WriteDataMessage *writeMsg = static_cast<WriteDataMessage *>(msg);
+    const QByteArray &ba = getByteArray(writeMsg->mimeData, writeMsg->mimeType);
+    writeMsg->data = ba;
+
+    return {writeMsg};
+}
+
+QString CollectWrittenDataCommand::name() const
+{
+    return QLatin1String("CollectWrittenDataCommand");
+}
+
+QList<CommandMessage *> WriteDataToFDCommand::doExecute(CommandMessage *msg)
+{
+    WriteDataMessage *writeMsg = static_cast<WriteDataMessage *>(msg);
+
+    QFile f;
+    if (f.open(writeMsg->fd, QFile::WriteOnly, QFile::AutoCloseHandle)) {
+        f.write(writeMsg->data);
+        f.close();
+    }
+
+    return {};
+}
+
+QString WriteDataToFDCommand::name() const
+{
+    return QLatin1String("WriteDataToFDCommand");
+}
+
+#define WRITEDATASERVICECOUNT 4
+
+QList<Command *> CollectWrittenDataService::commands()
+{
+    QList<Command *> commands;
+    for (int i = 0; i < WRITEDATASERVICECOUNT; ++i)
+        commands.append(new CollectWrittenDataCommand());
+
+    return commands;
+}
+
+QList<Command *> WriteDataToFDService::commands()
+{
+    QList<Command *> commands;
+    for (int i = 0; i < WRITEDATASERVICECOUNT; ++i)
+        commands.append(new WriteDataToFDCommand());
+
+    return commands;
+}
diff --git a/dde-clipboard-daemon/serviceflow/waylandcopyservice.h b/dde-clipboard-daemon/serviceflow/waylandcopyservice.h
new file mode 100644
index 0000000..940e304
--- /dev/null
+++ b/dde-clipboard-daemon/serviceflow/waylandcopyservice.h
@@ -0,0 +1,145 @@
+#ifndef WAYLANDCOPYSERVICE_H
+#define WAYLANDCOPYSERVICE_H
+
+#include "commandmessage.h"
+#include "command.h"
+#include "commandservice.h"
+
+#include <QMimeData>
+#include <QMutex>
+#include <QSharedPointer>
+
+namespace KWayland
+{
+namespace Client
+{
+class DataControlOfferV1;
+} //Client
+} //KWayland
+
+
+// for read data.
+class PipeCommandMessage : public CommandMessage
+{
+public:
+    KWayland::Client::DataControlOfferV1 *offer;
+    int pipeFds[2] = {0};
+    QMimeData *mimeData;
+    QStringList srcMimeTypes;
+    QStringList dstMimeTypes;
+};
+
+class CopyCommandMessage : public PipeCommandMessage
+{
+public:
+    int mimeTypeCount;
+    QString currentMimeType;
+    QByteArray data;
+};
+
+class MimeDataFilterCommand : public Command
+{
+public:
+    using Command::Command;
+    QList<CommandMessage *> doExecute(CommandMessage *msg) override;
+    QString name() const override;
+
+    QStringList filterMimeType(const QStringList &mimeTypeList);
+};
+
+class RequestReceiveCommand : public Command
+{
+public:
+    using Command::Command;
+    QList<CommandMessage *> doExecute(CommandMessage *msg) override;
+    QString name() const override;
+};
+
+class ReadDataCommand : public Command
+{
+public:
+    using Command::Command;
+    QList<CommandMessage *> doExecute(CommandMessage *msg) override;
+    QString name() const override;
+};
+
+class SyncMimeDataCommand : public Command
+{
+public:
+    using Command::Command;
+    QList<CommandMessage *> doExecute(CommandMessage *msg) override;
+    QString name() const override;
+
+private:
+    QAtomicInt m_mimeCount = 0;
+    QMutex m_mutex;
+};
+
+class MimeDataFilterService : public CommandService
+{
+public:
+    QList<Command *> commands() override;
+};
+
+class RequestReceiveService : public QObject, public CommandService
+{
+    Q_OBJECT
+
+public:
+    using QObject::QObject;
+    QList<Command *> commands() override;
+};
+
+class ReadDataService : public CommandService
+{
+public:
+    enum { MaxReadCommandCount = 8 };
+    QList<Command *> commands() override;
+};
+
+class SyncMimeDataService : public QObject, public CommandService
+{
+    Q_OBJECT
+public:
+    QList<Command *> commands() override;
+
+Q_SIGNALS:
+    void finished(bool success);
+};
+
+// for write data
+class WriteDataMessage : public CommandMessage
+{
+public:
+    QString mimeType;
+    QByteArray data;
+    QMimeData *mimeData;
+    qint32 fd;
+};
+
+class CollectWrittenDataCommand : public Command
+{
+public:
+    QList<CommandMessage *> doExecute(CommandMessage *);
+    QString name() const;
+};
+
+class WriteDataToFDCommand : public Command
+{
+public:
+    QList<CommandMessage *> doExecute(CommandMessage *);
+    QString name() const;
+};
+
+class CollectWrittenDataService : public CommandService
+{
+public:
+    QList<Command *> commands();
+};
+
+class WriteDataToFDService : public CommandService
+{
+public:
+    QList<Command *> commands();
+};
+#endif // WAYLANDCOPYSERVICE_H
diff --git a/dde-clipboard-daemon/waylandcopyclient.cpp b/dde-clipboard-daemon/waylandcopyclient.cpp
index 3fe0cbe..d3ca94e 100644
--- a/dde-clipboard-daemon/waylandcopyclient.cpp
+++ b/dde-clipboard-daemon/waylandcopyclient.cpp
@@ -1,17 +1,17 @@
-// SPDX-FileCopyrightText: 2011 - 2022 UnionTech Software Technology Co., Ltd.
+// SPDX-FileCopyrightText: 2011 - 2022 UnionTech Software Technology Co., Ltd.
 //
 // SPDX-License-Identifier: GPL-3.0-or-later
 
 #ifdef USE_DEEPIN_KF5_WAYLAND
 #include "waylandcopyclient.h"
-#include "readpipedatatask.h"
+#include "serviceflow/servicemanager.h"
+#include "serviceflow/waylandcopyservice.h"
 
-#include <QEventLoop>
-#include <QMimeData>
 #include <QImageReader>
-#include <QtConcurrent/QtConcurrent>
-#include <QImageWriter>
+#include <QMimeData>
 #include <QMutexLocker>
+#include <QThread>
+#include <QDebug>
 
 #include <KWayland/Client/connection_thread.h>
 #include <KWayland/Client/event_queue.h>
@@ -45,41 +45,7 @@ static inline QStringList imageReadMimeFormats()
     return imageMimeFormats(QImageReader::supportedImageFormats());
 }
 
-static QByteArray getByteArray(QMimeData *mimeData, const QString &mimeType)
-{
-    QByteArray content;
-    if (mimeType == QLatin1String("text/plain")) {
-        content = mimeData->text().toUtf8();
-    } else if (mimeData->hasImage()
-               && (mimeType == QLatin1String("application/x-qt-image")
-                   || mimeType.startsWith(QLatin1String("image/")))) {
-        QImage image = qvariant_cast<QImage>(mimeData->imageData());
-        if (!image.isNull()) {
-            QBuffer buf;
-            buf.open(QIODevice::ReadWrite);
-            QByteArray fmt = "BMP";
-            if (mimeType.startsWith(QLatin1String("image/"))) {
-                QByteArray imgFmt = mimeType.mid(6).toUpper().toLatin1();
-                if (QImageWriter::supportedImageFormats().contains(imgFmt))
-                    fmt = imgFmt;
-            }
-            QImageWriter wr(&buf, fmt);
-            wr.write(image);
-            content = buf.buffer();
-        }
-    } else if (mimeType == QLatin1String("application/x-color")) {
-        content = qvariant_cast<QColor>(mimeData->colorData()).name().toLatin1();
-    } else if (mimeType == QLatin1String("text/uri-list")) {
-        QList<QUrl> urls = mimeData->urls();
-        for (int i = 0; i < urls.count(); ++i) {
-            content.append(urls.at(i).toEncoded());
-            content.append('\n');
-        }
-    } else {
-        content = mimeData->data(mimeType);
-    }
-    return content;
-}
+
 
 DMimeData::DMimeData()
 {
@@ -139,9 +105,10 @@ WaylandCopyClient::WaylandCopyClient(QObject *parent)
     , m_copyControlSource(nullptr)
     , m_mimeData(new DMimeData())
     , m_seat(nullptr)
-    , m_curOffer(0)
+    , m_manager(new CommandServiceManager())
+    , m_registry(nullptr)
 {
-
+    init();
 }
 
 WaylandCopyClient::~WaylandCopyClient()
@@ -152,121 +119,141 @@ WaylandCopyClient::~WaylandCopyClient()
 
     if (m_mimeData)
         m_mimeData->deleteLater();
+
+    m_manager->unregisterService(m_writeServices[0]);
+    delete m_manager;
+    qDeleteAll(m_writeServices);
+    qDeleteAll(m_runningRootService);
 }
 
 void WaylandCopyClient::init()
 {
-    connect(m_connectionThreadObject, &ConnectionThread::connected, this, [this] {
-        m_eventQueue = new EventQueue(this);
-        m_eventQueue->setup(m_connectionThreadObject);
+    // alows 10 service flow to run.
+    m_manager->setMaxServiceCount(10);
 
-        Registry *registry = new Registry(this);
-        setupRegistry(registry);
-    }, Qt::QueuedConnection );
+    connect(m_connectionThreadObject, &ConnectionThread::connected, this, &WaylandCopyClient::onThreadConnected, Qt::UniqueConnection);
     m_connectionThreadObject->moveToThread(m_connectionThread);
     m_connectionThread->start();
     m_connectionThreadObject->initConnection();
-
-    // clipboard Manager的功能
-    connect(this, &WaylandCopyClient::dataChanged, this, &WaylandCopyClient::onDataChanged);
 }
 
-void WaylandCopyClient::setupRegistry(Registry *registry)
+void WaylandCopyClient::setupRegistry()
 {
-    connect(registry, &Registry::seatAnnounced, this, [this, registry] (quint32 name, quint32 version) {
-        m_seat = registry->createSeat(name, version, this);
-    });
+    if (m_registry)
+        delete m_registry;
 
-    connect(registry, &Registry::dataControlDeviceManagerAnnounced, this, [this, registry] (quint32 name, quint32 version) {
-        m_dataControlDeviceManager = registry->createDataControlDeviceManager(name, version, this);
-        m_dataControlDevice = m_dataControlDeviceManager->getDataDevice(m_seat, this);
+    m_registry = new Registry(this);
+    connect(m_registry, &Registry::seatAnnounced, this, &WaylandCopyClient::onRegistrySeatAnnounced, Qt::UniqueConnection);
+    connect(m_registry, &Registry::dataControlDeviceManagerAnnounced, this, &WaylandCopyClient::onDeviceManagerAnnounced, Qt::UniqueConnection);
 
-        connect(m_dataControlDevice, &DataControlDeviceV1::selectionCleared, this, [&] {
-                m_copyControlSource = nullptr;
-                sendOffer();
-        });
+    m_registry->setEventQueue(m_eventQueue);
+    m_registry->create(m_connectionThreadObject);
+    m_registry->setup();
+}
 
-        connect(m_dataControlDevice, &DataControlDeviceV1::dataOffered, this, &WaylandCopyClient::onDataOffered);
-    });
+void WaylandCopyClient::cleanServiceFlow(CommandService *service)
+{
+    auto root = m_manager->findRootService(service);
+    Q_ASSERT(root);
+
+    m_manager->unregisterService(root);
+    m_runningRootService.removeAll(root);
+    QList<CommandService *> services = {root};
+    CommandService *header = root;
+    while (auto next = header->nextService()) {
+        services.append(next);
+        header = next;
+    }
+    // clean service;
+    qDeleteAll(services);
+}
 
-    registry->setEventQueue(m_eventQueue);
-    registry->create(m_connectionThreadObject);
-    registry->setup();
+void WaylandCopyClient::wakePipeSyncCondition()
+{
+    m_pipeSyncCondition.wakeOne();
 }
 
-void WaylandCopyClient::onDataOffered(KWayland::Client::DataControlOfferV1* offer)
+void WaylandCopyClient::onThreadConnected()
 {
-    if (!offer)
-        return;
+    if (m_eventQueue)
+        delete m_eventQueue;
 
-    QList<QString> mimeTypeList = filterMimeType(offer->offeredMimeTypes());
-    if (mimeTypeList.isEmpty())
-        return;
+    m_eventQueue = new EventQueue(this);
+    m_eventQueue->setup(m_connectionThreadObject);
+    setupRegistry();
+}
 
-    m_curMimeTypes = mimeTypeList;
+void WaylandCopyClient::onDataOffered(KWayland::Client::DataControlOfferV1 *offer)
+{
+    MimeDataFilterService *mimefilterService = new MimeDataFilterService();
+    RequestReceiveService *requestReceiveService = new RequestReceiveService(this);
+    ReadDataService *readDataService = new ReadDataService();
+    SyncMimeDataService *syncMimeDataService = new SyncMimeDataService();
 
-    if (m_curOffer && qint64(offer) != m_curOffer) {
-        tryStopOldTask();
-    }
-    m_curOffer = qint64(offer);
+    mimefilterService->setNextService(requestReceiveService);
+    requestReceiveService->setNextService(readDataService);
+    readDataService->setNextService(syncMimeDataService);
 
-    if (!m_mimeData)
-        m_mimeData = new DMimeData();
-    m_mimeData->clear();
+    connect(syncMimeDataService, &SyncMimeDataService::finished, this, &WaylandCopyClient::onServiceFlowFinished);
 
-    execTask(mimeTypeList, offer);
+    m_manager->registerService(mimefilterService);
+    m_runningRootService.append(mimefilterService);
+    m_manager->start(mimefilterService);
+
+    PipeCommandMessage *msg = new PipeCommandMessage();
+    msg->offer = offer;
+    msg->mimeData = m_mimeData;
+    msg->srcMimeTypes = offer->offeredMimeTypes();
+    m_manager->appendFlowMessage(msg, mimefilterService);
+
+    QMutexLocker locker(&m_pipeSyncMutex);
+    m_pipeSyncCondition.wait(&m_pipeSyncMutex);
 }
 
-void WaylandCopyClient::execTask(const QStringList &mimeTypes, DataControlOfferV1 *offer)
+void WaylandCopyClient::onRegistrySeatAnnounced(quint32 name, quint32 version)
 {
-    QThreadPool *threadPool = QThreadPool::globalInstance();
-    if (!threadPool)
+    if (!m_registry->isValid())
         return;
-
-    for (const QString &mimeType : mimeTypes) {
-        ReadPipeDataTask *task = new ReadPipeDataTask(m_connectionThreadObject, offer, mimeType, this);
-        connect(task, &ReadPipeDataTask::dataReady, this, &WaylandCopyClient::taskDataReady);
-        task->setAutoDelete(true);
-        threadPool->start(task);
-
-        m_tasks.append(task);
-    }
+    m_seat = m_registry->createSeat(name, version, this);
 }
 
-void WaylandCopyClient::tryStopOldTask()
+void WaylandCopyClient::onDeviceManagerAnnounced(quint32 name, quint32 version)
 {
-    QThreadPool *threadPool = QThreadPool::globalInstance();
-    if (!threadPool)
+    if (m_dataControlDeviceManager)
+        m_dataControlDeviceManager->destroy();
+
+    m_dataControlDeviceManager = m_registry->createDataControlDeviceManager(name, version, this);
+    if (!m_dataControlDeviceManager->isValid())
         return;
 
-    for (ReadPipeDataTask *task : m_tasks) {
-        if (!threadPool->tryTake(task)){
-            if (task)
-                task->stopRunning();
+    auto dataControlDevice = m_dataControlDeviceManager->getDataDevice(m_seat, this);
+    if (dataControlDevice != m_dataControlDevice) {
+        m_dataControlDevice = dataControlDevice;
+
+        if (m_dataControlDevice) {
+            connect(m_dataControlDevice, &DataControlDeviceV1::selectionCleared, this, [&] {
+                    m_copyControlSource = nullptr;
+                    sendOffer();
+            });
+
+            connect(m_dataControlDevice, &DataControlDeviceV1::dataOffered, this, &WaylandCopyClient::onDataOffered);
         }
     }
 }
 
-void WaylandCopyClient::taskDataReady(qint64 offer, const QString &mimeType, const QByteArray &data)
+void WaylandCopyClient::onServiceFlowFinished(bool success)
 {
-    if (offer != m_curOffer) {
+    SyncMimeDataService *service = dynamic_cast<SyncMimeDataService *>(sender());
+    if (!service)
         return;
-    }
-
-    m_curMimeTypes.removeOne(mimeType);
-
-    m_mimeData->setData(mimeType, data);
-    if (m_curMimeTypes.isEmpty()) {
-        m_curOffer = 0;
-        m_tasks.clear();
 
+    if (success) {
+        // all data has updated.
         Q_EMIT dataChanged();
+        sendOffer();
     }
-}
 
-void WaylandCopyClient::onDataChanged()
-{
-    sendOffer();
+    cleanServiceFlow(service);
 }
 
 const QMimeData* WaylandCopyClient::mimeData()
@@ -287,11 +274,17 @@ void WaylandCopyClient::setMimeData(QMimeData *mimeData)
 
 void WaylandCopyClient::sendOffer()
 {
+    if (m_copyControlSource) {
+        m_copyControlSource->deleteLater();
+        disconnect(m_copyControlSource);
+    }
+
     m_copyControlSource = m_dataControlDeviceManager->createDataSource(this);
     if (!m_copyControlSource)
         return;
 
     connect(m_copyControlSource, &DataControlSourceV1::sendDataRequested, this, &WaylandCopyClient::onSendDataRequest);
+
     for (const QString &format : m_mimeData->formats()) {
         // 如果是application/x-qt-image类型则需要提供image的全部类型, 比如image/png
         if (ApplicationXQtImageLiteral == format) {
@@ -308,29 +301,25 @@ void WaylandCopyClient::sendOffer()
     m_connectionThreadObject->flush();
 }
 
-void WaylandCopyClient::onSendDataRequest(const QString &mimeType, qint32 fd) const
+void WaylandCopyClient::onSendDataRequest(const QString &mimeType, qint32 fd)
 {
-    QFile f;
-    if (f.open(fd, QFile::WriteOnly, QFile::AutoCloseHandle)) {
-        const QByteArray &ba = getByteArray(m_mimeData, mimeType);
-        f.write(ba);
-        f.close();
-    }
-}
+    if (m_writeServices.isEmpty()) {
+         CollectWrittenDataService *collectService = new CollectWrittenDataService();
+         WriteDataToFDService *wfdService = new WriteDataToFDService();
+         collectService->setNextService(wfdService);
 
-QStringList WaylandCopyClient::filterMimeType(const QStringList &mimeTypeList)
-{
-    QStringList tmpList;
-    for (const QString &mimeType : mimeTypeList) {
-        // 根据窗管的要求，不读取纯大写、和不含'/'的字段，因为源窗口可能没有写入这些字段的数据，导致获取数据的线程一直等待。
-        if ((mimeType.contains("/") && mimeType.toUpper() != mimeType)
-            || mimeType == "FROM_DEEPIN_CLIPBOARD_MANAGER"
-            || mimeType == "TIMESTAMP") {
-            tmpList.append(mimeType);
-        }
+         this->m_writeServices.insert(0, collectService);
+         this->m_writeServices.insert(1, wfdService);
+
+         this->m_manager->registerService(collectService);
+         this->m_manager->start(collectService);
     }
 
-    return tmpList;
+    WriteDataMessage *writeMsg = new WriteDataMessage();
+    writeMsg->fd = fd;
+    writeMsg->mimeType = mimeType;
+    writeMsg->mimeData = m_mimeData;
+    m_manager->appendFlowMessage(writeMsg, this->m_writeServices[0]);
 }
 
 #endif
diff --git a/dde-clipboard-daemon/waylandcopyclient.h b/dde-clipboard-daemon/waylandcopyclient.h
index 7464c78..fa0498c 100644
--- a/dde-clipboard-daemon/waylandcopyclient.h
+++ b/dde-clipboard-daemon/waylandcopyclient.h
@@ -2,6 +2,9 @@
 //
 // SPDX-License-Identifier: GPL-3.0-or-later
 
+#include <qobject.h>
+#include <qobjectdefs.h>
+#include <qrunnable.h>
 #ifdef USE_DEEPIN_KF5_WAYLAND
 #ifndef COPYCLIENT_H
 #define COPYCLIENT_H
@@ -10,6 +13,7 @@
 #include <QMimeData>
 #include <QPointer>
 #include <QMimeType>
+#include <QWaitCondition>
 
 namespace KWayland
 {
@@ -26,10 +30,9 @@ class DataControlOfferV1;
 } //Client
 } //KWayland
 
-class ReadPipeDataTask;
-
 using namespace KWayland::Client;
 
+
 class DMimeData : public QMimeData
 {
     Q_OBJECT
@@ -40,9 +43,12 @@ class DMimeData : public QMimeData
                                       QVariant::Type preferredType) const;
 };
 
+class CommandServiceManager;
+class CommandService;
 class WaylandCopyClient : public QObject
 {
     Q_OBJECT
+    friend class RequestReceiveCommand;
 
 public:
     explicit WaylandCopyClient(QObject *parent = nullptr);
@@ -52,23 +58,25 @@ class WaylandCopyClient : public QObject
     const QMimeData *mimeData();
     void setMimeData(QMimeData *mimeData);
     void sendOffer();
+    void call();
 
-private:
     void setupRegistry(Registry *registry);
+private:
+    void setupRegistry();
     QStringList filterMimeType(const QStringList &mimeTypeList);
+    void cleanServiceFlow(CommandService *service);
+    void wakePipeSyncCondition();
 
 Q_SIGNALS:
     void dataChanged();
 
 protected slots:
-    void onSendDataRequest(const QString &mimeType, qint32 fd) const;
+    void onThreadConnected();
+    void onSendDataRequest(const QString &mimeType, qint32 fd);
     void onDataOffered(DataControlOfferV1 *offer);
-    void onDataChanged();
-
-private:
-    void execTask(const QStringList &mimeTypes, DataControlOfferV1 *offer);
-    void tryStopOldTask();
-    void taskDataReady(qint64, const QString &mimeType, const QByteArray &data);
+    void onRegistrySeatAnnounced(quint32 name, quint32 version);
+    void onDeviceManagerAnnounced(quint32 name, quint32 version);
+    void onServiceFlowFinished(bool success);
 
 private:
     QThread *m_connectionThread;
@@ -79,10 +87,14 @@ protected slots:
     DataControlSourceV1 *m_copyControlSource;
     QPointer<QMimeData> m_mimeData;
     Seat *m_seat;
-
-    qint64 m_curOffer;
-    QStringList m_curMimeTypes;
-    QList<ReadPipeDataTask *> m_tasks;
+    CommandServiceManager *m_manager;
+    Registry *m_registry;
+    QList<CommandService *> m_runningRootService;
+    QMutex m_pipeSyncMutex;
+    // REMOVE IT: 窗管无法支持异步执行,因此使用服务流时需要额外的进行同步操作
+    // 否则可能出现访问野指针的风险.
+    QWaitCondition m_pipeSyncCondition;
+    QList<CommandService *> m_writeServices;
 };
 
 #endif // COPYCLIENT_H