From 20b52ff1ddc3b2f547b7081471f46dcfa5efabc7 Mon Sep 17 00:00:00 2001
From: Teemu Ollakka <teemu.ollakka@galeracluster.com>
Date: Mon, 11 Feb 2019 16:06:39 +0200
Subject: [PATCH] Allow direct manipulation of streaming context parameters.

Added a method to change streaming context fragment unit and
size. The method has a side effect of resetting unit counter.
---
 include/wsrep/client_state.hpp      | 10 ++++++++++
 include/wsrep/streaming_context.hpp | 25 +++++++++++++++++++++++++
 src/client_state.cpp                |  8 ++++++++
 3 files changed, 43 insertions(+)

diff --git a/include/wsrep/client_state.hpp b/include/wsrep/client_state.hpp
index 2f693f8..6d997fb 100644
--- a/include/wsrep/client_state.hpp
+++ b/include/wsrep/client_state.hpp
@@ -305,6 +305,16 @@ namespace wsrep
                     transaction_.after_row() : 0);
         }
 
+        /**
+         * Set streaming parameters.
+         *
+         * @param fragment_unit Desired fragment unit
+         * @param fragment_size Desired fragment size
+         */
+        void streaming_params(enum wsrep::streaming_context::fragment_unit
+                              fragment_unit,
+                              size_t fragment_size);
+
         /**
          * Enable streaming replication.
          *
diff --git a/include/wsrep/streaming_context.hpp b/include/wsrep/streaming_context.hpp
index a129f41..f1118aa 100644
--- a/include/wsrep/streaming_context.hpp
+++ b/include/wsrep/streaming_context.hpp
@@ -49,6 +49,31 @@ namespace wsrep
             , unit_counter_()
         { }
 
+        /**
+         * Set streaming parameters.
+         *
+         * Calling this method has a side effect of resetting unit
+         * counter.
+         *
+         * @param fragment_unit Desired fragment unit.
+         * @param fragment_size Desired fragment size.
+         */
+        void params(enum fragment_unit fragment_unit, size_t fragment_size)
+        {
+            if (fragment_size)
+            {
+                wsrep::log_debug() << "Enabling streaming: "
+                                   << fragment_unit << " " << fragment_size;
+            }
+            else
+            {
+                wsrep::log_debug() << "Disabling streaming";
+            }
+            fragment_unit_ = fragment_unit;
+            fragment_size_ = fragment_size;
+            reset_unit_counter();
+        }
+
         void enable(enum fragment_unit fragment_unit, size_t fragment_size)
         {
             wsrep::log_debug() << "Enabling streaming: "
diff --git a/src/client_state.cpp b/src/client_state.cpp
index 4c96b87..95e8a76 100644
--- a/src/client_state.cpp
+++ b/src/client_state.cpp
@@ -260,6 +260,14 @@ int wsrep::client_state::after_statement()
 //                             Streaming                                    //
 //////////////////////////////////////////////////////////////////////////////
 
+void wsrep::client_state::streaming_params(
+    enum wsrep::streaming_context::fragment_unit fragment_unit,
+    size_t fragment_size)
+{
+    assert(mode_ == m_local);
+    transaction_.streaming_context().params(fragment_unit, fragment_size);
+}
+
 int wsrep::client_state::enable_streaming(
     enum wsrep::streaming_context::fragment_unit
     fragment_unit,