From 641c0d12c9606326025b5544ce12a0c5828e67bf Mon Sep 17 00:00:00 2001 From: Yunsheng Bai Date: Tue, 7 Jun 2016 13:01:03 -0400 Subject: [PATCH] Mpdify Thrift: NOT tested because 'gradle run' not working --- .../ensemble/communication/ThriftHandler.java | 63 +++++++++++--- .../ensemble/communication/ThriftServer.java | 87 +++++++++++++------ 2 files changed, 112 insertions(+), 38 deletions(-) diff --git a/src/main/java/de/rwthaachen/ensemble/communication/ThriftHandler.java b/src/main/java/de/rwthaachen/ensemble/communication/ThriftHandler.java index 1d1e601..8dd2852 100644 --- a/src/main/java/de/rwthaachen/ensemble/communication/ThriftHandler.java +++ b/src/main/java/de/rwthaachen/ensemble/communication/ThriftHandler.java @@ -3,33 +3,74 @@ import de.rwthaachen.ensemble.model.AnswerCandidate; import de.rwthaachen.ensemble.orchestration.Controller; import de.rwthaachen.ensemble.orchestration.Pipeline; -import org.apache.thrift.TException; - +// Java packages import java.util.List; +import java.io.File; +import java.util.ArrayList; + +import org.apache.thrift.TException; +import org.apache.thrift.async.AsyncMethodCallback; -/** - * - */ -public class ThriftHandler extends Handler implements LucidaService.Iface { +public class ThriftHandler { + public static void print(String s) { + synchronized (System.out) { + System.out.println(s); + } + } + public static class SyncThriftHandler extends Handler implements LucidaService.Iface { Controller controller; - public ThriftHandler(Controller controller) { + public SyncThriftHandler(Controller controller) { this.controller = controller; } @Override public void create(String LUCID, QuerySpec spec) throws TException { - System.out.println("Create"); + print("Create"); } @Override public void learn(String LUCID, QuerySpec knowledge) throws TException { - System.out.println("learn"); + print("learn"); } @Override public String infer(String LUCID, QuerySpec query) throws TException { - System.out.println("Received question: " + LUCID); - return controller.getQaPipe().ask(LUCID); + if (query.content.isEmpty() || query.content.get(0).data.isEmpty()) { + throw new IllegalArgumentException(); + } + System.out.println("Received question: " + query.content.get(0).data.get(0)); + return controller.getQaPipe().ask(query.content.get(0).data.get(0)); + } + + public static class AsyncThriftHandler implements LucidaService.AsyncIface { + private SyncThriftHandler handler; + + public AsyncThriftHandler() { + handler = new SyncThriftHandler(); + } + + @Override + public void create(String LUCID, QuerySpec spec, AsyncMethodCallback resultHandler) + throws TException { + print("Async Create"); + handler.create(LUCID, spec); + resultHandler.onComplete(null); + } + + @Override + public void learn(String LUCID, QuerySpec knowledge, AsyncMethodCallback resultHandler) + throws TException { + print("Async Learn"); + handler.learn(LUCID, knowledge); + resultHandler.onComplete(null); + } + + @Override + public void infer(String LUCID, QuerySpec query, AsyncMethodCallback resultHandler) + throws TException { + print("Async Infer"); + resultHandler.onComplete(handler.infer(LUCID, query)); + } } } diff --git a/src/main/java/de/rwthaachen/ensemble/communication/ThriftServer.java b/src/main/java/de/rwthaachen/ensemble/communication/ThriftServer.java index 59078e6..985186b 100644 --- a/src/main/java/de/rwthaachen/ensemble/communication/ThriftServer.java +++ b/src/main/java/de/rwthaachen/ensemble/communication/ThriftServer.java @@ -2,38 +2,71 @@ import de.rwthaachen.ensemble.orchestration.Controller; import de.rwthaachen.ensemble.orchestration.Pipeline; -import org.apache.thrift.server.TServer; -import org.apache.thrift.server.TSimpleServer; -import org.apache.thrift.transport.TServerSocket; -import org.apache.thrift.transport.TServerTransport; -/** - * Created by fp on 5/13/16. - */ -public class ThriftServer { +// Thrift java libraries +import org.apache.thrift.server.TNonblockingServer; +import org.apache.thrift.transport.TFramedTransport; +import org.apache.thrift.transport.TNonblockingServerSocket; +import org.apache.thrift.transport.TNonblockingServerTransport; +import org.apache.thrift.server.TThreadedSelectorServer; +import org.apache.thrift.transport.TSocket; +import org.apache.thrift.transport.TTransport; +import org.apache.thrift.transport.TTransportException; - public static ThriftHandler thriftHandler; - public static LucidaService.Processor lucidaServiceProcessor; +import java.io.IOException; +import java.util.ArrayList; - public static void setup(Controller controller) { - thriftHandler = new ThriftHandler(controller); - lucidaServiceProcessor = new LucidaService.Processor(thriftHandler); - System.out.println("Created handler and processor."); - - Runnable simple = new Runnable() { - public void run() { simple(lucidaServiceProcessor); } - }; - new Thread(simple).start(); - } +import org.apache.thrift.TException; +import org.apache.thrift.TProcessor; +// Thrift client-side code for registering w/ sirius +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TProtocol; - public static void simple(LucidaService.Processor lucidaServiceProcessor) { +/** + * Starts the ensemble server and listens for requests. + */ +public class ThriftServer { + private static void connectToCMD() { + QueryInput query_input = new QueryInput(); + query_input.type = "ensemble"; + query_input.data = new ArrayList(); + query_input.data.add("localhost"); + query_input.tags = new ArrayList(); + query_input.tags.add("9090"); + QuerySpec spec = new QuerySpec(); + spec.content = new ArrayList(); + spec.content.add(query_input); + // Initialize thrift objects. + TTransport transport = new TSocket("localhost", 8080); + TProtocol protocol = new TBinaryProtocol(new TFramedTransport(transport)); + LucidaService.Client client = new LucidaService.Client(protocol); try { - TServerTransport serverTransport = new TServerSocket(9090); - TServer server = new TSimpleServer(new TServer.Args(serverTransport).processor(lucidaServiceProcessor)); - System.out.println("Starting server..."); - server.serve(); - } catch (Exception e) { - e.printStackTrace(); + transport.open(); + System.out.println("Connecting to CMD at port " + 8080); + // Register itself to CMD. + client.create("", spec); + transport.close(); + System.out.println("Successfully connected to CMD"); + } catch (TException x) { + x.printStackTrace(); } } + + /** + * Entry point for ensemble. + */ + public static void setup(Controller controller) + throws TTransportException, IOException, InterruptedException { + connectToCMD(); + TProcessor proc = new LucidaService.AsyncProcessor( + new ThriftHandler.AsyncThriftHandler(controller)); + TNonblockingServerTransport transport = new TNonblockingServerSocket(9090); + TThreadedSelectorServer.Args arguments = new TThreadedSelectorServer.Args(transport) + .processor(proc) + .protocolFactory(new TBinaryProtocol.Factory()) + .transportFactory(new TFramedTransport.Factory()); + final TThreadedSelectorServer server = new TThreadedSelectorServer(arguments); + System.out.println("ensemble at port 9090"); + server.serve(); + } }