server: Optimize sending for `SubscriptionSink::pipe_from_stream` (#901)

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>

core/src/server/rpc_module.rs

@@ -886,13 +886,7 @@ impl SubscriptionSink {
 			return Ok(false);
 		}
 
-		// Only possible to trigger when the connection is dropped.
-		if self.is_closed() {
-			return Ok(false);
-		}
-
-		let msg = self.build_message(result)?;
-		Ok(self.inner.send_raw(msg).is_ok())
+		self.send_without_accept(result)
 	}
 
 	/// Reads data from the `stream` and sends back data on the subscription
@@ -976,7 +970,7 @@ impl SubscriptionSink {
 			match futures_util::future::select(stream_item, closed_fut).await {
 				// The app sent us a value to send back to the subscribers
 				Either::Left((Ok(Some(result)), next_closed_fut)) => {
-					match self.send(&result) {
+					match self.send_without_accept(&result) {
 						Ok(true) => (),
 						Ok(false) => {
 							break SubscriptionClosed::RemotePeerAborted;
@@ -1034,6 +1028,21 @@ impl SubscriptionSink {
 		self.inner.is_closed() || self.close_notify.is_none() || !self.is_active_subscription()
 	}
 
+	/// Send a message back to subscribers.
+	///
+	/// This is similar to the [`SubscriptionSink::send`], but it does not try to accept
+	/// the subscription prior to sending.
+	#[inline]
+	fn send_without_accept<T: Serialize>(&mut self, result: &T) -> Result<bool, serde_json::Error> {
+		// Only possible to trigger when the connection is dropped.
+		if self.is_closed() {
+			return Ok(false);
+		}
+
+		let msg = self.build_message(result)?;
+		Ok(self.inner.send_raw(msg).is_ok())
+	}
+
 	fn is_active_subscription(&self) -> bool {
 		match self.unsubscribe.as_ref() {
 			Some(unsubscribe) => unsubscribe.has_changed().is_ok(),