lib/rs/src/server/multiplexed.rs - packaging/sources/thrift - Gitiles

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements. See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership. The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License. You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied. See the License for the
 // specific language governing permissions and limitations
 // under the License.

 use log::debug;

 use std::collections::HashMap;
 use std::convert::Into;
 use std::fmt;
 use std::fmt::{Debug, Formatter};
 use std::sync::{Arc, Mutex};

 use crate::protocol::{TInputProtocol, TMessageIdentifier, TOutputProtocol, TStoredInputProtocol};

 use super::{handle_process_result, TProcessor};

 const MISSING_SEPARATOR_AND_NO_DEFAULT: &str =
     "missing service separator and no default processor set";
 type ThreadSafeProcessor = Box<dyn TProcessor + Send + Sync>;

 /// A `TProcessor` that can demux service calls to multiple underlying
 /// Thrift services.
 ///
 /// Users register service-specific `TProcessor` instances with a
 /// `TMultiplexedProcessor`, and then register that processor with a server
 /// implementation. Following that, all incoming service calls are automatically
 /// routed to the service-specific `TProcessor`.
 ///
 /// A `TMultiplexedProcessor` can only handle messages sent by a
 /// `TMultiplexedOutputProtocol`.
 #[derive(Default)]
 pub struct TMultiplexedProcessor {
     stored: Mutex<StoredProcessors>,
 }

 #[derive(Default)]
 struct StoredProcessors {
     processors: HashMap<String, Arc<ThreadSafeProcessor>>,
     default_processor: Option<Arc<ThreadSafeProcessor>>,
 }

 impl TMultiplexedProcessor {
     /// Create a new `TMultiplexedProcessor` with no registered service-specific
     /// processors.
     pub fn new() -> TMultiplexedProcessor {
         TMultiplexedProcessor {
             stored: Mutex::new(StoredProcessors {
                 processors: HashMap::new(),
                 default_processor: None,
             }),
         }
     }

     /// Register a service-specific `processor` for the service named
     /// `service_name`. This implementation is also backwards-compatible with
     /// non-multiplexed clients. Set `as_default` to `true` to allow
     /// non-namespaced requests to be dispatched to a default processor.
     ///
     /// Returns success if a new entry was inserted. Returns an error if:
     /// * A processor exists for `service_name`
     /// * You attempt to register a processor as default, and an existing default exists
     #[allow(clippy::map_entry)]
     pub fn register<S: Into<String>>(
         &mut self,
         service_name: S,
         processor: Box<dyn TProcessor + Send + Sync>,
         as_default: bool,
     ) -> crate::Result<()> {
         let mut stored = self.stored.lock().unwrap();

         let name = service_name.into();
         if !stored.processors.contains_key(&name) {
             let processor = Arc::new(processor);

             if as_default {
                 if stored.default_processor.is_none() {
                     stored.processors.insert(name, processor.clone());
                     stored.default_processor = Some(processor.clone());
                     Ok(())
                 } else {
                     Err("cannot reset default processor".into())
                 }
             } else {
                 stored.processors.insert(name, processor);
                 Ok(())
             }
         } else {
             Err(format!("cannot overwrite existing processor for service {}", name).into())
         }
     }

     fn process_message(
         &self,
         msg_ident: &TMessageIdentifier,
         i_prot: &mut dyn TInputProtocol,
         o_prot: &mut dyn TOutputProtocol,
     ) -> crate::Result<()> {
         let (svc_name, svc_call) = split_ident_name(&msg_ident.name);
         debug!("routing svc_name {:?} svc_call {}", &svc_name, &svc_call);

         let processor: Option<Arc<ThreadSafeProcessor>> = {
             let stored = self.stored.lock().unwrap();
             if let Some(name) = svc_name {
                 stored.processors.get(name).cloned()
             } else {
                 stored.default_processor.clone()
             }
         };

         match processor {
             Some(arc) => {
                 let new_msg_ident = TMessageIdentifier::new(
                     svc_call,
                     msg_ident.message_type,
                     msg_ident.sequence_number,
                 );
                 let mut proxy_i_prot = TStoredInputProtocol::new(i_prot, new_msg_ident);
                 (*arc).process(&mut proxy_i_prot, o_prot)
             }
             None => Err(missing_processor_message(svc_name).into()),
         }
     }
 }

 impl TProcessor for TMultiplexedProcessor {
     fn process(&self, i_prot: &mut dyn TInputProtocol, o_prot: &mut dyn TOutputProtocol) -> crate::Result<()> {
         let msg_ident = i_prot.read_message_begin()?;

         debug!("process incoming msg id:{:?}", &msg_ident);
         let res = self.process_message(&msg_ident, i_prot, o_prot);

         handle_process_result(&msg_ident, res, o_prot)
     }
 }

 impl Debug for TMultiplexedProcessor {
     fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
         let stored = self.stored.lock().unwrap();
         write!(
             f,
             "TMultiplexedProcess {{ registered_count: {:?} default: {:?} }}",
             stored.processors.keys().len(),
             stored.default_processor.is_some()
         )
     }
 }

 fn split_ident_name(ident_name: &str) -> (Option<&str>, &str) {
     ident_name
         .find(':')
         .map(|pos| {
             let (svc_name, svc_call) = ident_name.split_at(pos);
             let (_, svc_call) = svc_call.split_at(1); // remove colon from service call name
             (Some(svc_name), svc_call)
         })
         .or_else(|| Some((None, ident_name)))
         .unwrap()
 }

 fn missing_processor_message(svc_name: Option<&str>) -> String {
     match svc_name {
         Some(name) => format!("no processor found for service {}", name),
         None => MISSING_SEPARATOR_AND_NO_DEFAULT.to_owned(),
     }
 }

 #[cfg(test)]
 mod tests {
     use std::convert::Into;
     use std::sync::atomic::{AtomicBool, Ordering};
     use std::sync::Arc;

     use crate::protocol::{TBinaryInputProtocol, TBinaryOutputProtocol, TMessageIdentifier, TMessageType};
     use crate::transport::{ReadHalf, TBufferChannel, TIoChannel, WriteHalf};
     use crate::{ApplicationError, ApplicationErrorKind};

     use super::*;

     #[test]
     fn should_split_name_into_proper_separator_and_service_call() {
         let ident_name = "foo:bar_call";
         let (serv, call) = split_ident_name(&ident_name);
         assert_eq!(serv, Some("foo"));
         assert_eq!(call, "bar_call");
     }

     #[test]
     fn should_return_full_ident_if_no_separator_exists() {
         let ident_name = "bar_call";
         let (serv, call) = split_ident_name(&ident_name);
         assert_eq!(serv, None);
         assert_eq!(call, "bar_call");
     }

     #[test]
     fn should_write_error_if_no_separator_found_and_no_default_processor_exists() {
         let (mut i, mut o) = build_objects();

         let sent_ident = TMessageIdentifier::new("foo", TMessageType::Call, 10);
         o.write_message_begin(&sent_ident).unwrap();
         o.flush().unwrap();
         o.transport.copy_write_buffer_to_read_buffer();
         o.transport.empty_write_buffer();

         let p = TMultiplexedProcessor::new();
         p.process(&mut i, &mut o).unwrap(); // at this point an error should be written out

         i.transport.set_readable_bytes(&o.transport.write_bytes());
         let rcvd_ident = i.read_message_begin().unwrap();
         let expected_ident = TMessageIdentifier::new("foo", TMessageType::Exception, 10);
         assert_eq!(rcvd_ident, expected_ident);
         let rcvd_err = crate::Error::read_application_error_from_in_protocol(&mut i).unwrap();
         let expected_err = ApplicationError::new(
             ApplicationErrorKind::Unknown,
             MISSING_SEPARATOR_AND_NO_DEFAULT,
         );
         assert_eq!(rcvd_err, expected_err);
     }

     #[test]
     fn should_write_error_if_separator_exists_and_no_processor_found() {
         let (mut i, mut o) = build_objects();

         let sent_ident = TMessageIdentifier::new("missing:call", TMessageType::Call, 10);
         o.write_message_begin(&sent_ident).unwrap();
         o.flush().unwrap();
         o.transport.copy_write_buffer_to_read_buffer();
         o.transport.empty_write_buffer();

         let p = TMultiplexedProcessor::new();
         p.process(&mut i, &mut o).unwrap(); // at this point an error should be written out

         i.transport.set_readable_bytes(&o.transport.write_bytes());
         let rcvd_ident = i.read_message_begin().unwrap();
         let expected_ident = TMessageIdentifier::new("missing:call", TMessageType::Exception, 10);
         assert_eq!(rcvd_ident, expected_ident);
         let rcvd_err = crate::Error::read_application_error_from_in_protocol(&mut i).unwrap();
         let expected_err = ApplicationError::new(
             ApplicationErrorKind::Unknown,
             missing_processor_message(Some("missing")),
         );
         assert_eq!(rcvd_err, expected_err);
     }

     #[derive(Default)]
     struct Service {
         pub invoked: Arc<AtomicBool>,
     }

     impl TProcessor for Service {
         fn process(&self, _: &mut dyn TInputProtocol, _: &mut dyn TOutputProtocol) -> crate::Result<()> {
             let res = self
                 .invoked
                 .compare_and_swap(false, true, Ordering::Relaxed);
             if res {
                 Ok(())
             } else {
                 Err("failed swap".into())
             }
         }
     }

     #[test]
     fn should_route_call_to_correct_processor() {
         let (mut i, mut o) = build_objects();

         // build the services
         let svc_1 = Service {
             invoked: Arc::new(AtomicBool::new(false)),
         };
         let atm_1 = svc_1.invoked.clone();
         let svc_2 = Service {
             invoked: Arc::new(AtomicBool::new(false)),
         };
         let atm_2 = svc_2.invoked.clone();

         // register them
         let mut p = TMultiplexedProcessor::new();
         p.register("service_1", Box::new(svc_1), false).unwrap();
         p.register("service_2", Box::new(svc_2), false).unwrap();

         // make the service call
         let sent_ident = TMessageIdentifier::new("service_1:call", TMessageType::Call, 10);
         o.write_message_begin(&sent_ident).unwrap();
         o.flush().unwrap();
         o.transport.copy_write_buffer_to_read_buffer();
         o.transport.empty_write_buffer();

         p.process(&mut i, &mut o).unwrap();

         // service 1 should have been invoked, not service 2
         assert_eq!(atm_1.load(Ordering::Relaxed), true);
         assert_eq!(atm_2.load(Ordering::Relaxed), false);
     }

     #[test]
     fn should_route_call_to_correct_processor_if_no_separator_exists_and_default_processor_set() {
         let (mut i, mut o) = build_objects();

         // build the services
         let svc_1 = Service {
             invoked: Arc::new(AtomicBool::new(false)),
         };
         let atm_1 = svc_1.invoked.clone();
         let svc_2 = Service {
             invoked: Arc::new(AtomicBool::new(false)),
         };
         let atm_2 = svc_2.invoked.clone();

         // register them
         let mut p = TMultiplexedProcessor::new();
         p.register("service_1", Box::new(svc_1), false).unwrap();
         p.register("service_2", Box::new(svc_2), true).unwrap(); // second processor is default

         // make the service call (it's an old client, so we have to be backwards compatible)
         let sent_ident = TMessageIdentifier::new("old_call", TMessageType::Call, 10);
         o.write_message_begin(&sent_ident).unwrap();
         o.flush().unwrap();
         o.transport.copy_write_buffer_to_read_buffer();
         o.transport.empty_write_buffer();

         p.process(&mut i, &mut o).unwrap();

         // service 2 should have been invoked, not service 1
         assert_eq!(atm_1.load(Ordering::Relaxed), false);
         assert_eq!(atm_2.load(Ordering::Relaxed), true);
     }

     fn build_objects() -> (
         TBinaryInputProtocol<ReadHalf<TBufferChannel>>,
         TBinaryOutputProtocol<WriteHalf<TBufferChannel>>,
     ) {
         let c = TBufferChannel::with_capacity(128, 128);
         let (r_c, w_c) = c.split().unwrap();
         (
             TBinaryInputProtocol::new(r_c, true),
             TBinaryOutputProtocol::new(w_c, true),
         )
     }
 }
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing,
	// software distributed under the License is distributed on an
	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	// KIND, either express or implied. See the License for the
	// specific language governing permissions and limitations
	// under the License.

	use log::debug;

	use std::collections::HashMap;
	use std::convert::Into;
	use std::fmt;
	use std::fmt::{Debug, Formatter};
	use std::sync::{Arc, Mutex};

	use crate::protocol::{TInputProtocol, TMessageIdentifier, TOutputProtocol, TStoredInputProtocol};

	use super::{handle_process_result, TProcessor};

	const MISSING_SEPARATOR_AND_NO_DEFAULT: &str =
	"missing service separator and no default processor set";
	type ThreadSafeProcessor = Box<dyn TProcessor + Send + Sync>;

	/// A `TProcessor` that can demux service calls to multiple underlying
	/// Thrift services.
	///
	/// Users register service-specific `TProcessor` instances with a
	/// `TMultiplexedProcessor`, and then register that processor with a server
	/// implementation. Following that, all incoming service calls are automatically
	/// routed to the service-specific `TProcessor`.
	///
	/// A `TMultiplexedProcessor` can only handle messages sent by a
	/// `TMultiplexedOutputProtocol`.
	#[derive(Default)]
	pub struct TMultiplexedProcessor {
	stored: Mutex<StoredProcessors>,
	}

	#[derive(Default)]
	struct StoredProcessors {
	processors: HashMap<String, Arc<ThreadSafeProcessor>>,
	default_processor: Option<Arc<ThreadSafeProcessor>>,
	}

	impl TMultiplexedProcessor {
	/// Create a new `TMultiplexedProcessor` with no registered service-specific
	/// processors.
	pub fn new() -> TMultiplexedProcessor {
	TMultiplexedProcessor {
	stored: Mutex::new(StoredProcessors {
	processors: HashMap::new(),
	default_processor: None,
	}),
	}
	}

	/// Register a service-specific `processor` for the service named
	/// `service_name`. This implementation is also backwards-compatible with
	/// non-multiplexed clients. Set `as_default` to `true` to allow
	/// non-namespaced requests to be dispatched to a default processor.
	///
	/// Returns success if a new entry was inserted. Returns an error if:
	/// * A processor exists for `service_name`
	/// * You attempt to register a processor as default, and an existing default exists
	#[allow(clippy::map_entry)]
	pub fn register<S: Into<String>>(
	&mut self,
	service_name: S,
	processor: Box<dyn TProcessor + Send + Sync>,
	as_default: bool,
	) -> crate::Result<()> {
	let mut stored = self.stored.lock().unwrap();

	let name = service_name.into();
	if !stored.processors.contains_key(&name) {
	let processor = Arc::new(processor);

	if as_default {
	if stored.default_processor.is_none() {
	stored.processors.insert(name, processor.clone());
	stored.default_processor = Some(processor.clone());
	Ok(())
	} else {
	Err("cannot reset default processor".into())
	}
	} else {
	stored.processors.insert(name, processor);
	Ok(())
	}
	} else {
	Err(format!("cannot overwrite existing processor for service {}", name).into())
	}
	}

	fn process_message(
	&self,
	msg_ident: &TMessageIdentifier,
	i_prot: &mut dyn TInputProtocol,
	o_prot: &mut dyn TOutputProtocol,
	) -> crate::Result<()> {
	let (svc_name, svc_call) = split_ident_name(&msg_ident.name);
	debug!("routing svc_name {:?} svc_call {}", &svc_name, &svc_call);

	let processor: Option<Arc<ThreadSafeProcessor>> = {
	let stored = self.stored.lock().unwrap();
	if let Some(name) = svc_name {
	stored.processors.get(name).cloned()
	} else {
	stored.default_processor.clone()
	}
	};

	match processor {
	Some(arc) => {
	let new_msg_ident = TMessageIdentifier::new(
	svc_call,
	msg_ident.message_type,
	msg_ident.sequence_number,
	);
	let mut proxy_i_prot = TStoredInputProtocol::new(i_prot, new_msg_ident);
	(*arc).process(&mut proxy_i_prot, o_prot)
	}
	None => Err(missing_processor_message(svc_name).into()),
	}
	}
	}

	impl TProcessor for TMultiplexedProcessor {
	fn process(&self, i_prot: &mut dyn TInputProtocol, o_prot: &mut dyn TOutputProtocol) -> crate::Result<()> {
	let msg_ident = i_prot.read_message_begin()?;

	debug!("process incoming msg id:{:?}", &msg_ident);
	let res = self.process_message(&msg_ident, i_prot, o_prot);

	handle_process_result(&msg_ident, res, o_prot)
	}
	}

	impl Debug for TMultiplexedProcessor {
	fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
	let stored = self.stored.lock().unwrap();
	write!(
	f,
	"TMultiplexedProcess {{ registered_count: {:?} default: {:?} }}",
	stored.processors.keys().len(),
	stored.default_processor.is_some()
	)
	}
	}

	fn split_ident_name(ident_name: &str) -> (Option<&str>, &str) {
	ident_name
	.find(':')
	.map(\|pos\| {
	let (svc_name, svc_call) = ident_name.split_at(pos);
	let (_, svc_call) = svc_call.split_at(1); // remove colon from service call name
	(Some(svc_name), svc_call)
	})
	.or_else(\|\| Some((None, ident_name)))
	.unwrap()
	}

	fn missing_processor_message(svc_name: Option<&str>) -> String {
	match svc_name {
	Some(name) => format!("no processor found for service {}", name),
	None => MISSING_SEPARATOR_AND_NO_DEFAULT.to_owned(),
	}
	}

	#[cfg(test)]
	mod tests {
	use std::convert::Into;
	use std::sync::atomic::{AtomicBool, Ordering};
	use std::sync::Arc;

	use crate::protocol::{TBinaryInputProtocol, TBinaryOutputProtocol, TMessageIdentifier, TMessageType};
	use crate::transport::{ReadHalf, TBufferChannel, TIoChannel, WriteHalf};
	use crate::{ApplicationError, ApplicationErrorKind};

	use super::*;

	#[test]
	fn should_split_name_into_proper_separator_and_service_call() {
	let ident_name = "foo:bar_call";
	let (serv, call) = split_ident_name(&ident_name);
	assert_eq!(serv, Some("foo"));
	assert_eq!(call, "bar_call");
	}

	#[test]
	fn should_return_full_ident_if_no_separator_exists() {
	let ident_name = "bar_call";
	let (serv, call) = split_ident_name(&ident_name);
	assert_eq!(serv, None);
	assert_eq!(call, "bar_call");
	}

	#[test]
	fn should_write_error_if_no_separator_found_and_no_default_processor_exists() {
	let (mut i, mut o) = build_objects();

	let sent_ident = TMessageIdentifier::new("foo", TMessageType::Call, 10);
	o.write_message_begin(&sent_ident).unwrap();
	o.flush().unwrap();
	o.transport.copy_write_buffer_to_read_buffer();
	o.transport.empty_write_buffer();

	let p = TMultiplexedProcessor::new();
	p.process(&mut i, &mut o).unwrap(); // at this point an error should be written out

	i.transport.set_readable_bytes(&o.transport.write_bytes());
	let rcvd_ident = i.read_message_begin().unwrap();
	let expected_ident = TMessageIdentifier::new("foo", TMessageType::Exception, 10);
	assert_eq!(rcvd_ident, expected_ident);
	let rcvd_err = crate::Error::read_application_error_from_in_protocol(&mut i).unwrap();
	let expected_err = ApplicationError::new(
	ApplicationErrorKind::Unknown,
	MISSING_SEPARATOR_AND_NO_DEFAULT,
	);
	assert_eq!(rcvd_err, expected_err);
	}

	#[test]
	fn should_write_error_if_separator_exists_and_no_processor_found() {
	let (mut i, mut o) = build_objects();

	let sent_ident = TMessageIdentifier::new("missing:call", TMessageType::Call, 10);
	o.write_message_begin(&sent_ident).unwrap();
	o.flush().unwrap();
	o.transport.copy_write_buffer_to_read_buffer();
	o.transport.empty_write_buffer();

	let p = TMultiplexedProcessor::new();
	p.process(&mut i, &mut o).unwrap(); // at this point an error should be written out

	i.transport.set_readable_bytes(&o.transport.write_bytes());
	let rcvd_ident = i.read_message_begin().unwrap();
	let expected_ident = TMessageIdentifier::new("missing:call", TMessageType::Exception, 10);
	assert_eq!(rcvd_ident, expected_ident);
	let rcvd_err = crate::Error::read_application_error_from_in_protocol(&mut i).unwrap();
	let expected_err = ApplicationError::new(
	ApplicationErrorKind::Unknown,
	missing_processor_message(Some("missing")),
	);
	assert_eq!(rcvd_err, expected_err);
	}

	#[derive(Default)]
	struct Service {
	pub invoked: Arc<AtomicBool>,
	}

	impl TProcessor for Service {
	fn process(&self, _: &mut dyn TInputProtocol, _: &mut dyn TOutputProtocol) -> crate::Result<()> {
	let res = self
	.invoked
	.compare_and_swap(false, true, Ordering::Relaxed);
	if res {
	Ok(())
	} else {
	Err("failed swap".into())
	}
	}
	}

	#[test]
	fn should_route_call_to_correct_processor() {
	let (mut i, mut o) = build_objects();

	// build the services
	let svc_1 = Service {
	invoked: Arc::new(AtomicBool::new(false)),
	};
	let atm_1 = svc_1.invoked.clone();
	let svc_2 = Service {
	invoked: Arc::new(AtomicBool::new(false)),
	};
	let atm_2 = svc_2.invoked.clone();

	// register them
	let mut p = TMultiplexedProcessor::new();
	p.register("service_1", Box::new(svc_1), false).unwrap();
	p.register("service_2", Box::new(svc_2), false).unwrap();

	// make the service call
	let sent_ident = TMessageIdentifier::new("service_1:call", TMessageType::Call, 10);
	o.write_message_begin(&sent_ident).unwrap();
	o.flush().unwrap();
	o.transport.copy_write_buffer_to_read_buffer();
	o.transport.empty_write_buffer();

	p.process(&mut i, &mut o).unwrap();

	// service 1 should have been invoked, not service 2
	assert_eq!(atm_1.load(Ordering::Relaxed), true);
	assert_eq!(atm_2.load(Ordering::Relaxed), false);
	}

	#[test]
	fn should_route_call_to_correct_processor_if_no_separator_exists_and_default_processor_set() {
	let (mut i, mut o) = build_objects();

	// build the services
	let svc_1 = Service {
	invoked: Arc::new(AtomicBool::new(false)),
	};
	let atm_1 = svc_1.invoked.clone();
	let svc_2 = Service {
	invoked: Arc::new(AtomicBool::new(false)),
	};
	let atm_2 = svc_2.invoked.clone();

	// register them
	let mut p = TMultiplexedProcessor::new();
	p.register("service_1", Box::new(svc_1), false).unwrap();
	p.register("service_2", Box::new(svc_2), true).unwrap(); // second processor is default

	// make the service call (it's an old client, so we have to be backwards compatible)
	let sent_ident = TMessageIdentifier::new("old_call", TMessageType::Call, 10);
	o.write_message_begin(&sent_ident).unwrap();
	o.flush().unwrap();
	o.transport.copy_write_buffer_to_read_buffer();
	o.transport.empty_write_buffer();

	p.process(&mut i, &mut o).unwrap();

	// service 2 should have been invoked, not service 1
	assert_eq!(atm_1.load(Ordering::Relaxed), false);
	assert_eq!(atm_2.load(Ordering::Relaxed), true);
	}

	fn build_objects() -> (
	TBinaryInputProtocol<ReadHalf<TBufferChannel>>,
	TBinaryOutputProtocol<WriteHalf<TBufferChannel>>,
	) {
	let c = TBufferChannel::with_capacity(128, 128);
	let (r_c, w_c) = c.split().unwrap();
	(
	TBinaryInputProtocol::new(r_c, true),
	TBinaryOutputProtocol::new(w_c, true),
	)
	}
	}