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

@@ -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.
          *

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: "

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,