Add method client_state::stream()

This method forces a streaming replication step.

include/wsrep/client_state.hpp

@@ -354,6 +354,14 @@ namespace wsrep
          */
         int after_row();
 
+        /**
+         * Force a streaming step
+         *
+         * This method can be used to replicate a fragment in the
+         * current context.
+         */
+        int stream();
+
         /**
          * Set streaming parameters.
          *

include/wsrep/transaction.hpp

@@ -175,6 +175,8 @@ namespace wsrep
 
         int after_row();
 
+        int stream();
+
         int before_prepare(wsrep::unique_lock<wsrep::mutex>&);
 
         int after_prepare(wsrep::unique_lock<wsrep::mutex>&);

src/client_state.cpp

@@ -336,6 +336,13 @@ int wsrep::client_state::after_row()
                 : 0);
 }
 
+int wsrep::client_state::stream()
+{
+    assert(mode_ == m_local);
+    assert(state_ == s_exec);
+    return transaction_.stream();
+}
+
 void wsrep::client_state::fragment_applied(wsrep::seqno seqno)
 {
     assert(mode_ == m_high_priority);

src/transaction.cpp

@@ -273,6 +273,15 @@ int wsrep::transaction::after_row()
     return ret;
 }
 
+int wsrep::transaction::stream()
+{
+    wsrep::unique_lock<wsrep::mutex> lock(client_state_.mutex());
+    debug_log_state("stream_enter");
+    int ret(streaming_step(lock, true));
+    debug_log_state("stream_leave");
+    return ret;
+}
+
 int wsrep::transaction::before_prepare(
     wsrep::unique_lock<wsrep::mutex>& lock)
 {