FunctionCall { Object apply(Object[] args, TProtocol protocol, int seqid) throws Exception { return apply(args, protocol, seqid, null); } FunctionCall(Method method, Thrift thrift); boolean getIsObservable(Method method);  }
@Test public void shouldReceiveObserable() throws Exception { final TestService.Processor<Iface> processor = new TestService.Processor<Iface>(new Iface() { @Override public void notify(int id) throws TException { } @Override public UnionB get(int id) throws TestException, TException { return new UnionB(UnionB._Fields.OS, new com.meituan.firefly.testthrift.OrderedStruct(1)); } @Override public List<UnionB> getList(List<Short> ids) throws TException { return null; } @Override public void notifyWithoutOneway(int id) throws TestException, TException { } @Override public UnionB emptyArgMethod() throws TestException, TException { return null; } }); TTransport transport = new FlushableMemoryBuffer(4096) { boolean flushed = false; @Override public void flush() throws TTransportException { if (!flushed) { flushed = true; try { processor.process(new TBinaryProtocol(this), new TBinaryProtocol(this)); } catch (TException e) { throw new RuntimeException(e); } } } }; TestScheduler testScheduler = new TestScheduler(); FunctionCall functionCall = new FunctionCall(com.meituan.firefly.rx_testfirefly.TestService.class.getMethod("get", Integer.class), thrift); Observable<com.meituan.firefly.rx_testfirefly.UnionB> observable; try { observable = (Observable<com.meituan.firefly.rx_testfirefly.UnionB>) functionCall.apply(new Object[]{1}, new TBinaryProtocol(transport), 1, testScheduler); } catch (ClassCastException e) { throw new ClassCastException("return type of method is not obserable !"); } Assert.assertNotNull(observable); observable.subscribeOn(testScheduler); observable.observeOn(testScheduler); TestSubscriber<com.meituan.firefly.rx_testfirefly.UnionB> testSubscriber = new TestSubscriber(); observable.subscribe(testSubscriber); testScheduler.triggerActions(); Assert.assertEquals(1, testSubscriber.getOnNextEvents().size()); com.meituan.firefly.rx_testfirefly.UnionB u = testSubscriber.getOnNextEvents().get(0); Assert.assertNotNull(u); Assert.assertEquals((Integer) 1, u.os.id); }
@Test public void emptyArgMethod() throws Exception { final TestService.Processor<Iface> processor = new TestService.Processor<Iface>(new Iface() { @Override public void notify(int id) throws TException { } @Override public UnionB get(int id) throws TestException, TException { return new UnionB(UnionB._Fields.OS, new com.meituan.firefly.testthrift.OrderedStruct(1)); } @Override public List<UnionB> getList(List<Short> ids) throws TException { return null; } @Override public void notifyWithoutOneway(int id) throws TestException, TException { } @Override public UnionB emptyArgMethod() throws TestException, TException { return new UnionB(UnionB._Fields.OS, new com.meituan.firefly.testthrift.OrderedStruct(1)); } }); TTransport transport = new FlushableMemoryBuffer(4096) { boolean flushed = false; @Override public void flush() throws TTransportException { if (!flushed) { flushed = true; try { processor.process(new TBinaryProtocol(this), new TBinaryProtocol(this)); } catch (TException e) { throw new RuntimeException(e); } } } }; TestScheduler testScheduler = new TestScheduler(); FunctionCall functionCall = new FunctionCall(com.meituan.firefly.rx_testfirefly.TestService.class.getMethod("emptyArgMethod", null), thrift); Observable<com.meituan.firefly.rx_testfirefly.UnionB> observable; try { observable = (Observable<com.meituan.firefly.rx_testfirefly.UnionB>) functionCall.apply(null, new TBinaryProtocol(transport), 1, testScheduler); } catch (ClassCastException e) { throw new ClassCastException("return type of method is not obserable !"); } Assert.assertNotNull(observable); observable.subscribeOn(testScheduler); observable.observeOn(testScheduler); TestSubscriber<com.meituan.firefly.rx_testfirefly.UnionB> testSubscriber = new TestSubscriber(); observable.subscribe(testSubscriber); testScheduler.triggerActions(); Assert.assertEquals(1, testSubscriber.getOnNextEvents().size()); com.meituan.firefly.rx_testfirefly.UnionB u = testSubscriber.getOnNextEvents().get(0); Assert.assertNotNull(u); Assert.assertEquals((Integer) 1, u.os.id); }
@Test public void emptyArgMethodReceiveObserable() throws Exception { final TestService.Processor<Iface> processor = new TestService.Processor<Iface>(new Iface() { @Override public void notify(int id) throws TException { } @Override public UnionB get(int id) throws TestException, TException { return null; } @Override public List<UnionB> getList(List<Short> ids) throws TException { return null; } @Override public void notifyWithoutOneway(int id) throws TestException, TException { } @Override public UnionB emptyArgMethod() throws TestException, TException { return new UnionB(UnionB._Fields.OS, new com.meituan.firefly.testthrift.OrderedStruct(1)); } }); TTransport transport = new FlushableMemoryBuffer(4096) { boolean flushed = false; @Override public void flush() throws TTransportException { if (!flushed) { flushed = true; try { processor.process(new TBinaryProtocol(this), new TBinaryProtocol(this)); } catch (TException e) { throw new RuntimeException(e); } } } }; TestScheduler testScheduler = new TestScheduler(); FunctionCall functionCall = new FunctionCall(com.meituan.firefly.rx_testfirefly.TestService.class.getMethod("emptyArgMethod", null), thrift); Observable<com.meituan.firefly.rx_testfirefly.UnionB> observable; try { observable = (Observable<com.meituan.firefly.rx_testfirefly.UnionB>) functionCall.apply(null, new TBinaryProtocol(transport), 1, testScheduler); } catch (ClassCastException e) { throw new ClassCastException("return type of method is not obserable !"); } Assert.assertNotNull(observable); observable.subscribeOn(testScheduler); observable.observeOn(testScheduler); TestSubscriber<com.meituan.firefly.rx_testfirefly.UnionB> testSubscriber = new TestSubscriber(); observable.subscribe(testSubscriber); testScheduler.triggerActions(); Assert.assertEquals(1, testSubscriber.getOnNextEvents().size()); com.meituan.firefly.rx_testfirefly.UnionB u = testSubscriber.getOnNextEvents().get(0); Assert.assertNotNull(u); Assert.assertEquals((Integer) 1, u.os.id); }