1 // Licensed to the Apache Software Foundation (ASF) under one
2 // or more contributor license agreements. See the NOTICE file
3 // distributed with this work for additional information
4 // regarding copyright ownership. The ASF licenses this file
5 // to you under the Apache License, Version 2.0 (the
6 // "License"); you may not use this file except in compliance
7 // with the License. You may obtain a copy of the License at
8 //
9 // http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing,
12 // software distributed under the License is distributed on an
13 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14 // KIND, either express or implied. See the License for the
15 // specific language governing permissions and limitations
16 // under the License.
17
18 use std::net::{TcpListener, TcpStream};
19 use std::sync::Arc;
20 use threadpool::ThreadPool;
21
22 use protocol::{TInputProtocol, TInputProtocolFactory, TOutputProtocol, TOutputProtocolFactory};
23 use transport::{TIoChannel, TReadTransportFactory, TTcpChannel, TWriteTransportFactory};
24 use {ApplicationError, ApplicationErrorKind};
25
26 use super::TProcessor;
27
28 /// Fixed-size thread-pool blocking Thrift server.
29 ///
30 /// A `TServer` listens on a given address and submits accepted connections
31 /// to an **unbounded** queue. Connections from this queue are serviced by
32 /// the first available worker thread from a **fixed-size** thread pool. Each
33 /// accepted connection is handled by that worker thread, and communication
34 /// over this thread occurs sequentially and synchronously (i.e. calls block).
35 /// Accepted connections have an input half and an output half, each of which
36 /// uses a `TTransport` and `TInputProtocol`/`TOutputProtocol` to translate
37 /// messages to and from byes. Any combination of `TInputProtocol`, `TOutputProtocol`
38 /// and `TTransport` may be used.
39 ///
40 /// # Examples
41 ///
42 /// Creating and running a `TServer` using Thrift-compiler-generated
43 /// service code.
44 ///
45 /// ```no_run
46 /// use thrift;
47 /// use thrift::protocol::{TInputProtocolFactory, TOutputProtocolFactory};
48 /// use thrift::protocol::{TBinaryInputProtocolFactory, TBinaryOutputProtocolFactory};
49 /// use thrift::protocol::{TInputProtocol, TOutputProtocol};
50 /// use thrift::transport::{TBufferedReadTransportFactory, TBufferedWriteTransportFactory,
51 /// TReadTransportFactory, TWriteTransportFactory};
52 /// use thrift::server::{TProcessor, TServer};
53 ///
54 /// //
55 /// // auto-generated
56 /// //
57 ///
58 /// // processor for `SimpleService`
59 /// struct SimpleServiceSyncProcessor;
60 /// impl SimpleServiceSyncProcessor {
61 /// fn new<H: SimpleServiceSyncHandler>(processor: H) -> SimpleServiceSyncProcessor {
62 /// unimplemented!();
63 /// }
64 /// }
65 ///
66 /// // `TProcessor` implementation for `SimpleService`
67 /// impl TProcessor for SimpleServiceSyncProcessor {
68 /// fn process(&self, i: &mut TInputProtocol, o: &mut TOutputProtocol) -> thrift::Result<()> {
69 /// unimplemented!();
70 /// }
71 /// }
72 ///
73 /// // service functions for SimpleService
74 /// trait SimpleServiceSyncHandler {
75 /// fn service_call(&self) -> thrift::Result<()>;
76 /// }
77 ///
78 /// //
79 /// // user-code follows
80 /// //
81 ///
82 /// // define a handler that will be invoked when `service_call` is received
83 /// struct SimpleServiceHandlerImpl;
84 /// impl SimpleServiceSyncHandler for SimpleServiceHandlerImpl {
85 /// fn service_call(&self) -> thrift::Result<()> {
86 /// unimplemented!();
87 /// }
88 /// }
89 ///
90 /// // instantiate the processor
91 /// let processor = SimpleServiceSyncProcessor::new(SimpleServiceHandlerImpl {});
92 ///
93 /// // instantiate the server
94 /// let i_tr_fact: Box<TReadTransportFactory> = Box::new(TBufferedReadTransportFactory::new());
95 /// let i_pr_fact: Box<TInputProtocolFactory> = Box::new(TBinaryInputProtocolFactory::new());
96 /// let o_tr_fact: Box<TWriteTransportFactory> = Box::new(TBufferedWriteTransportFactory::new());
97 /// let o_pr_fact: Box<TOutputProtocolFactory> = Box::new(TBinaryOutputProtocolFactory::new());
98 ///
99 /// let mut server = TServer::new(
100 /// i_tr_fact,
101 /// i_pr_fact,
102 /// o_tr_fact,
103 /// o_pr_fact,
104 /// processor,
105 /// 10
106 /// );
107 ///
108 /// // start listening for incoming connections
109 /// match server.listen("127.0.0.1:8080") {
110 /// Ok(_) => println!("listen completed"),
111 /// Err(e) => println!("listen failed with error {:?}", e),
112 /// }
113 /// ```
114 #[derive(Debug)]
115 pub struct TServer<PRC, RTF, IPF, WTF, OPF>
116 where
117 PRC: TProcessor + Send + Sync + 'static,
118 RTF: TReadTransportFactory + 'static,
119 IPF: TInputProtocolFactory + 'static,
120 WTF: TWriteTransportFactory + 'static,
121 OPF: TOutputProtocolFactory + 'static,
122 {
123 r_trans_factory: RTF,
124 i_proto_factory: IPF,
125 w_trans_factory: WTF,
126 o_proto_factory: OPF,
127 processor: Arc<PRC>,
128 worker_pool: ThreadPool,
129 }
130
131 impl<PRC, RTF, IPF, WTF, OPF> TServer<PRC, RTF, IPF, WTF, OPF>
132 where
133 PRC: TProcessor + Send + Sync + 'static,
134 RTF: TReadTransportFactory + 'static,
135 IPF: TInputProtocolFactory + 'static,
136 WTF: TWriteTransportFactory + 'static,
137 OPF: TOutputProtocolFactory + 'static,
138 {
139 /// Create a `TServer`.
140 ///
141 /// Each accepted connection has an input and output half, each of which
142 /// requires a `TTransport` and `TProtocol`. `TServer` uses
143 /// `read_transport_factory` and `input_protocol_factory` to create
144 /// implementations for the input, and `write_transport_factory` and
145 /// `output_protocol_factory` to create implementations for the output.
new( read_transport_factory: RTF, input_protocol_factory: IPF, write_transport_factory: WTF, output_protocol_factory: OPF, processor: PRC, num_workers: usize, ) -> TServer<PRC, RTF, IPF, WTF, OPF>146 pub fn new(
147 read_transport_factory: RTF,
148 input_protocol_factory: IPF,
149 write_transport_factory: WTF,
150 output_protocol_factory: OPF,
151 processor: PRC,
152 num_workers: usize,
153 ) -> TServer<PRC, RTF, IPF, WTF, OPF> {
154 TServer {
155 r_trans_factory: read_transport_factory,
156 i_proto_factory: input_protocol_factory,
157 w_trans_factory: write_transport_factory,
158 o_proto_factory: output_protocol_factory,
159 processor: Arc::new(processor),
160 worker_pool: ThreadPool::with_name("Thrift service processor".to_owned(), num_workers),
161 }
162 }
163
164 /// Listen for incoming connections on `listen_address`.
165 ///
166 /// `listen_address` should be in the form `host:port`,
167 /// for example: `127.0.0.1:8080`.
168 ///
169 /// Return `()` if successful.
170 ///
171 /// Return `Err` when the server cannot bind to `listen_address` or there
172 /// is an unrecoverable error.
listen(&mut self, listen_address: &str) -> ::Result<()>173 pub fn listen(&mut self, listen_address: &str) -> ::Result<()> {
174 let listener = TcpListener::bind(listen_address)?;
175 for stream in listener.incoming() {
176 match stream {
177 Ok(s) => {
178 let (i_prot, o_prot) = self.new_protocols_for_connection(s)?;
179 let processor = self.processor.clone();
180 self.worker_pool
181 .execute(move || handle_incoming_connection(processor, i_prot, o_prot));
182 }
183 Err(e) => {
184 warn!("failed to accept remote connection with error {:?}", e);
185 }
186 }
187 }
188
189 Err(::Error::Application(ApplicationError {
190 kind: ApplicationErrorKind::Unknown,
191 message: "aborted listen loop".into(),
192 }))
193 }
194
new_protocols_for_connection( &mut self, stream: TcpStream, ) -> ::Result<(Box<TInputProtocol + Send>, Box<TOutputProtocol + Send>)>195 fn new_protocols_for_connection(
196 &mut self,
197 stream: TcpStream,
198 ) -> ::Result<(Box<TInputProtocol + Send>, Box<TOutputProtocol + Send>)> {
199 // create the shared tcp stream
200 let channel = TTcpChannel::with_stream(stream);
201
202 // split it into two - one to be owned by the
203 // input tran/proto and the other by the output
204 let (r_chan, w_chan) = channel.split()?;
205
206 // input protocol and transport
207 let r_tran = self.r_trans_factory.create(Box::new(r_chan));
208 let i_prot = self.i_proto_factory.create(r_tran);
209
210 // output protocol and transport
211 let w_tran = self.w_trans_factory.create(Box::new(w_chan));
212 let o_prot = self.o_proto_factory.create(w_tran);
213
214 Ok((i_prot, o_prot))
215 }
216 }
217
handle_incoming_connection<PRC>( processor: Arc<PRC>, i_prot: Box<TInputProtocol>, o_prot: Box<TOutputProtocol>, ) where PRC: TProcessor,218 fn handle_incoming_connection<PRC>(
219 processor: Arc<PRC>,
220 i_prot: Box<TInputProtocol>,
221 o_prot: Box<TOutputProtocol>,
222 ) where
223 PRC: TProcessor,
224 {
225 let mut i_prot = i_prot;
226 let mut o_prot = o_prot;
227 loop {
228 let r = processor.process(&mut *i_prot, &mut *o_prot);
229 if let Err(e) = r {
230 warn!("processor completed with error: {:?}", e);
231 break;
232 }
233 }
234 }
235