Lifetime failures

compile_error

allen@mrwiggles ~/s/r/t/l/rs> cargo build
   Compiling thrift v1.0.0 (file:///Users/allen/src/rust_projects/thrift/lib/rs)
error[E0495]: cannot infer an appropriate lifetime for borrow expression due to conflicting requirements
   --> src/server/threaded.rs:160:29
    |
160 |                     let p = &mut self.processor; // FIXME: why can't I inline this?
    |                             ^^^^^^^^^^^^^^^^^^^
    |
note: first, the lifetime cannot outlive the anonymous lifetime #1 defined on the body at 154:67...
   --> src/server/threaded.rs:154:68
    |
154 |     pub fn listen(&mut self, listen_address: &str) -> ::Result<()> {
    |                                                                    ^
note: ...so that reference does not outlive borrowed content
   --> src/server/threaded.rs:160:29
    |
160 |                     let p = &mut self.processor; // FIXME: why can't I inline this?
    |                             ^^^^^^^^^^^^^^^^^^^
    = note: but, the lifetime must be valid for the static lifetime...
note: ...so that the type `[closure@src/server/threaded.rs:161:48: 161:101 p:&mut PR, i_prot:std::boxed::Box<protocol::TInputProtocol>, o_prot:std::boxed::Box<protocol::TOutputProtocol>]` will meet its required lifetime bounds
   --> src/server/threaded.rs:161:34
    |
161 |                     let handle = thread::spawn(move || handle_incoming_connection(p, i_prot, o_prot));
    |                                  ^^^^^^^^^^^^^

error: aborting due to previous error

error: Could not compile `thrift`.

threaded.rs

// 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 std::cell::RefCell;
use std::net::{TcpListener, TcpStream};
use std::rc::Rc;
use std::thread;
use std::thread::JoinHandle;

use ::{ApplicationError, ApplicationErrorKind};
use ::protocol::{TInputProtocol, TInputProtocolFactory, TOutputProtocol, TOutputProtocolFactory};
use ::transport::{TTcpChannel, TIOChannel, TTransport, TTransportFactory};

use super::TProcessor;

pub struct TThreadedServer<PR: TProcessor> {
    i_trans_factory: Box<TTransportFactory>,
    i_proto_factory: Box<TInputProtocolFactory>,
    o_trans_factory: Box<TTransportFactory>,
    o_proto_factory: Box<TOutputProtocolFactory>,
    processing_threads: Vec<JoinHandle<()>>,
    processor: PR,
}

impl<PR: TProcessor + Send + Sync> TThreadedServer<PR> {
    pub fn new(input_transport_factory: Box<TTransportFactory>,
               input_protocol_factory: Box<TInputProtocolFactory>,
               output_transport_factory: Box<TTransportFactory>,
               output_protocol_factory: Box<TOutputProtocolFactory>,
               processor: PR)
               -> TThreadedServer<PR> {
        TThreadedServer {
            i_trans_factory: input_transport_factory,
            i_proto_factory: input_protocol_factory,
            o_trans_factory: output_transport_factory,
            o_proto_factory: output_protocol_factory,
            processing_threads: Vec::new(),
            processor: processor,
        }
    }

    pub fn listen(&mut self, listen_address: &str) -> ::Result<()> {
        let listener = TcpListener::bind(listen_address)?;
        for stream in listener.incoming() {
            match stream {
                Ok(s) => {
                    let (i_prot, o_prot) = self.new_protocols_for_connection(s)?;
                    let p = &mut self.processor; // FIXME: <--- ERROR HAPPENS HERE
                    let handle = thread::spawn(move || handle_incoming_connection(p, i_prot, o_prot));
                },
                Err(e) => {
                    warn!("failed to accept remote connection with error {:?}", e);
                },
            }
        }

        Err(::Error::Application(ApplicationError {
            kind: ApplicationErrorKind::Unknown,
            message: "aborted listen loop".into(),
        }))
    }


    fn new_protocols_for_connection(&mut self, stream: TcpStream) -> ::Result<(Box<TInputProtocol>, Box<TOutputProtocol>)>{
        // create the shared tcp stream
        let channel = TTcpChannel::with_stream(stream);
        let (i_chan, o_chan) = channel.try_clone()?;

        // input protocol and transport
        let i_tran = self.i_trans_factory.create(i_chan);
        let i_prot = self.i_proto_factory.create(i_tran);

        // output protocol and transport
        let o_tran = self.o_trans_factory.create(o_chan);
        let o_prot = self.o_proto_factory.create(o_tran);

        Ok((i_prot, o_prot))
    }
}

impl <PR: TProcessor + Send + Sync> Drop for TThreadedServer<PR> {
    fn drop(&mut self) {
        for h in &self.processing_threads {
            if !h.join().is_ok() {
                println!("failed to terminate and join processing thread {:?}", h.thread())
            }
        }

        self.processing_threads.clear();
    }
}

fn handle_incoming_connection<PR>(processor: &mut PR, i_prot: Box<TInputProtocol>, o_prot: Box<TOutputProtocol>) where PR: TProcessor {
    loop {
        let r = processor.process(&mut *i_prot, &mut *o_prot);
        if let Err(e) = r {
            warn!("processor failed with error: {:?}", e);
            break; // FIXME: close here
        }
    }
}