Spring Thrift, Hessian like over Http

 

public class ThriftExporter extends RemoteExporter implements InitializingBean {

	private static final Logger LOGGER = LoggerFactory.getLogger(ThriftExporter.class);

	public static final String CONTENT_TYPE_HESSIAN = "application/x-thrift";

	protected TProcessorFactory processorFactory_;

	protected TTransportFactory inputTransportFactory_ = new TTransportFactory();
	protected TTransportFactory outputTransportFactory_ = new TTransportFactory();

	protected TProtocolFactory inputProtocolFactory_;
	protected TProtocolFactory outputProtocolFactory_;

	protected TServerEventHandler eventHandler_;

	protected Class<?> processorClass;

	@Override
	public void afterPropertiesSet() throws Exception {
		//		LocationThrfitTestService.Processor<LocationThrfitTestService.Iface> processor = new LocationThrfitTestService.Processor<LocationThrfitTestService.Iface>(new LocationThrfitTestServiceImpl());
		//		LocationThrfitTestService.Processor<LocationThrfitTestService.Iface> processor = new LocationThrfitTestService.Processor<LocationThrfitTestService.Iface>((LocationThrfitTestService.Iface) getProxyForService());

		Object service = getService();
		Class<?> serviceInterface = getServiceInterface();
		Constructor<?> constructor = processorClass.getConstructor(serviceInterface);
		TProcessor processor = (TProcessor) constructor.newInstance(getProxyForService());

		processorFactory_ = new TProcessorFactory(processor);

		TBinaryProtocol.Factory portFactory = new TBinaryProtocol.Factory(true, true);
		inputProtocolFactory_ = portFactory;
		outputProtocolFactory_ = portFactory;

		eventHandler_ = null;
	}

	public void invoke(InputStream inputStream, OutputStream outputStream) throws Throwable {
		//Assert.notNull(this.skeleton, "Thrift exporter has not been initialized");
		//doInvoke(this.skeleton, inputStream, outputStream);

		TIOStreamTransport client_ = new TIOStreamTransport(inputStream, outputStream);

		TProcessor processor = null;
		TTransport inputTransport = null;
		TTransport outputTransport = null;
		TProtocol inputProtocol = null;
		TProtocol outputProtocol = null;

		TServerEventHandler eventHandler = null;
		ServerContext connectionContext = null;

		try {
			processor = processorFactory_.getProcessor(client_);
			inputTransport = inputTransportFactory_.getTransport(client_);
			outputTransport = outputTransportFactory_.getTransport(client_);
			inputProtocol = inputProtocolFactory_.getProtocol(inputTransport);
			outputProtocol = outputProtocolFactory_.getProtocol(outputTransport);	  

			eventHandler = getEventHandler();
			if (eventHandler != null) {
				connectionContext = eventHandler.createContext(inputProtocol, outputProtocol);
			}
			// we check stopped_ first to make sure we're not supposed to be shutting
			// down. this is necessary for graceful shutdown.
			// while (true) {
			//
			//     if (eventHandler != null) {
			//         eventHandler.processContext(connectionContext, inputTransport, outputTransport);
			//     }
			//
			//     if(stopped_ || !processor.process(inputProtocol, outputProtocol)) {
			//         break;
			//     }
			// }


			if (eventHandler != null) {
				eventHandler.processContext(connectionContext, inputTransport, outputTransport);
			}

			processor.process(inputProtocol, outputProtocol);
		} catch (TSaslTransportException ttx) {
			// Something thats not SASL was in the stream, continue silently 
		} catch (TTransportException ttx) {
			// Assume the client died and continue silently
		} catch (TException tx) {
			LOGGER.error("Thrift error occurred during processing of message.", tx);
		} catch (Exception x) {
			LOGGER.error("Error occurred during processing of message.", x);
		} finally {
			if (eventHandler != null) {
				eventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
			}
			if (inputTransport != null) {
				inputTransport.close();
			}
			if (outputTransport != null) {
				outputTransport.close();
			}
			if (client_.isOpen()) {
				client_.close();
			}
		}
	}

	public TServerEventHandler getEventHandler() {
		return eventHandler_;
	}

	public void setProcessorClass(Class<?> processorClass) {
		this.processorClass = processorClass;
	}
}

 

 

 

public class ThriftServiceExporter extends ThriftExporter implements HttpRequestHandler {

	/**
	 * 
	 */
	@Override
	public void handleRequest(HttpServletRequest request, HttpServletResponse response)
			throws ServletException, IOException {
		if (!"POST".equals(request.getMethod())) {
			throw new HttpRequestMethodNotSupportedException(request.getMethod(),
					new String[] {"POST"}, "ThriftServiceExporter only supports POST requests");
		}

		response.setContentType(CONTENT_TYPE_HESSIAN);
		try {
		  invoke(request.getInputStream(), response.getOutputStream());
		}
		catch (Throwable ex) {
		  throw new NestedServletException("Thrift skeleton invocation failed", ex);
		}
	}

}

 

public class ThriftProxy implements InvocationHandler {
	private static final Logger log = Logger.getLogger(ThriftProxy.class.getName());
        ....

        protected Class<?> skeletonClass;

        /**
	 * Package protected constructor for factory
	 */
	ThriftProxy(ThriftProxyFactory factory, URL url)
	{
		_factory = factory;
		_url = url;
	}

	/**
	 * Protected constructor for subclassing
	 */
	protected ThriftProxy(URL url, ThriftProxyFactory factory)
	{
		_factory = factory;
		_url = url;
	}

	/**
	 * Returns the proxy's URL.
	 */
	public URL getURL()
	{
		return _url;
	}

	/**
	 * Handles the object invocation.
	 *
	 * @param proxy the proxy object to invoke
	 * @param method the method to call
	 * @param args the arguments to the proxy object
	 */
	public Object invoke(Object proxy, Method method, Object []args) throws Throwable {
		THttpClient transport = new THttpClient(_url.toString());
		TProtocol protocol = new TBinaryProtocol(transport);

		Constructor<?> constructor = skeletonClass.getConstructor(new Class<?>[] {TProtocol.class});
		proxy = constructor.newInstance(protocol);
		Method targetMethod = skeletonClass.getMethod(method.getName(), method.getParameterTypes());
		return targetMethod.invoke(proxy, args);
	}
	
	public void setSkeletonClass(Class<?> skeletonClass) {
		this.skeletonClass = skeletonClass;
	}

        ....

}

 

public class ThriftProxyFactory implements ServiceProxyFactory, ObjectFactory {

    .......
  
    protected Class<?> skeletonClass;
	
	public ThriftProxyFactory() {
		
	}
	
	public ThriftProxyFactory(Class<?> skeletonClass) {
		this.skeletonClass = skeletonClass;
	}

	@Override
	public Object create(Class api, String url) throws MalformedURLException {
		return create(api, url,
				Thread.currentThread().getContextClassLoader());
	}

	public Object create(Class api, String urlName, ClassLoader loader)
			throws MalformedURLException
	{
		if (api == null)
			throw new NullPointerException("api must not be null for HessianProxyFactory.create()");
		ThriftProxy handler = null;

		if (false && urlName.startsWith("jms:")) {
			/*
		      String jndiName = urlName.substring("jms:".length());

		      try {
		        handler = new HessianJMSProxy(this, jndiName, _connectionFactoryName);
		      } catch (Exception e) {
		        log.info("Unable to create JMS proxy: " + e);
		        return null;
		      }
			 */
		} 
		else {
			URL url = new URL(urlName); 
			handler = new ThriftProxy(this, url);
			handler.setSkeletonClass(skeletonClass);
		}

		return Proxy.newProxyInstance(loader,
				new Class[] {api},
				handler);
	}

	public void setSkeletonClass(Class<?> skeletonClass) {
		this.skeletonClass = skeletonClass;
	}

      ....
}

 

 

namespace java com.chos.test.service

# >thrift-0.10.0.exe --gen java -out . ./com/chos/test/service/LocationThrfitTestService.thrift

struct Location {
    1: optional i32 id;
    
    # 经度
    2: optional double longitude,
    
    # 纬度
    3: optional double latitude,
    
    # 海拔高度
    4: optional double altitude
}

service LocationThrfitTestService {
    void add(Location location), 
    Location get(i32 id), 
    list<Location> getList()
}

 

 

 

@Service("locationThrfitTestService")
public class LocationThrfitTestServiceImpl implements LocationThrfitTestService.Iface {

	private Map<Integer, Location> list;
	
	public LocationThrfitTestServiceImpl() {
		list = new ConcurrentHashMap<>();
	}
	
	@Override
	public void add(Location location) throws TException {
		list.put(location.getId(), location);
	}

	@Override
	public Location get(int id) throws TException {
		return list.get(id);
	}

	@Override
	public List<Location> getList() throws TException {
		return new LinkedList<>(list.values());
	}

}

 

 

<bean name="/LocationThrfitTestService"    
        class="com.chos.test.ThriftServiceExporter">    
        <!-- service的ref与HelloServiceImpl中@Service中配置的一致 -->    
        <property name="service" ref="locationThrfitTestService" />    
        <property name="processorClass" 
            value="com.chos.test.service.LocationThrfitTestService.Processor" />
        <!-- 接口的路径 -->
        <property name="serviceInterface"    
            value="com.chos.test.service.LocationThrfitTestService.Iface" />    
    </bean>

 

    <servlet>  
    	<servlet-name>remote</servlet-name>  
    	<servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>  
    	<load-on-startup>1</load-on-startup>  
	</servlet>  
  
	<servlet-mapping>  
    	<servlet-name>remote</servlet-name>  
    	<url-pattern>/service/*</url-pattern>  
	</servlet-mapping>

 

 

@Test
public void add() {
	try {
		String url = "http://localhost:8080/springhessiantest/service/LocationThrfitTestService";    
		
		// com.chos.test.ThriftProxyFactory instead
		ThriftProxyFactory factory = new ThriftProxyFactory(LocationThrfitTestService.Client.class);  
		LocationThrfitTestService.Iface locationThrfitTestService = (LocationThrfitTestService.Iface) factory.create(LocationThrfitTestService.Iface.class, url);    
		
		Location location = new Location();
		location.setId(3);
        location.setLongitude(5109.5);
        location.setLatitude(712920.317);
        location.setAltitude(100);
		locationThrfitTestService.add(location);    
		
		Location ret = locationThrfitTestService.get(location.getId());
		System.out.println(ret.toString());
	} catch (Exception e) {    
		e.printStackTrace();    
	}
}

@Test
public void getList() {
	try {
		String url = "http://localhost:8080/springhessiantest/service/LocationThrfitTestService";    
		
		// com.chos.test.ThriftProxyFactory instead
		ThriftProxyFactory factory = new ThriftProxyFactory(LocationThrfitTestService.Client.class);  
		LocationThrfitTestService.Iface locationThrfitTestService = (LocationThrfitTestService.Iface) factory.create(LocationThrfitTestService.Iface.class, url);    
		
        List<Location> list = locationThrfitTestService.getList();
        for (Location location : list) {
        	System.out.println(location.toString());
        }
	} catch (Exception e) {    
		e.printStackTrace();    
	}
}