package reactmann;

import com.aphyr.riemann.Proto;
import io.vertx.rxcore.java.RxVertx;
import io.vertx.rxcore.java.http.RxHttpServer;
import io.vertx.rxcore.java.net.RxNetServer;
import org.apache.http.NameValuePair;
import org.apache.http.client.utils.URLEncodedUtils;
import org.vertx.java.core.buffer.Buffer;
import org.vertx.java.core.json.JsonArray;
import org.vertx.java.core.json.JsonObject;
import org.vertx.java.core.streams.WriteStream;
import org.vertx.java.platform.Verticle;

import java.net.URI;
import java.util.List;

public class TcpMessageVerticle extends Verticle {

   public void start() {
      RxVertx rxVertx = new RxVertx(vertx);
      RxNetServer netServer = rxVertx.createNetServer();
      RxHttpServer httpServer = rxVertx.createHttpServer();

      httpServer.websocket().flatMap(s -> {
         try {
            List<NameValuePair> query = URLEncodedUtils.parse(new URI(s.uri()), "UTF-8");
            NameValuePair nameValuePair = query.stream().filter(p -> "query".equals(p.getName())).findAny().get();

            return Riemann.getEvents(vertx)
               .filter(Query.parse(nameValuePair.getValue()))
               .map(e -> new JsonObject()
                  .putArray("tags", new JsonArray(e.getTags().toArray()))
                  .putString("host", e.getHost())
                  .putString("state", e.getState())
                  .putString("service", e.getService())
                  .putString("description", e.getDescription())
                  .putNumber("metric", e.getMetric())
                  .putNumber("time", e.getTime())
                  .putNumber("ttl", e.getTtl()))
               .map(j -> Tup2.create(s, j));

         } catch (Exception e) {
            throw new NetSocketException(s, e);
         }
      }).subscribe(m -> m.getLeft().writeTextFrame(m.getRight().encode()));

      netServer
         .connectStream()
         .flatMap(s -> Riemann.convertBufferStreamToMessages(s.coreSocket(), s.asObservable()))
         .subscribe(s -> {
            sendResponse(Proto.Msg.newBuilder().setOk(true).build(), s.getLeft());
            vertx.eventBus().publish("riemann.stream", s.getRight().toByteArray());
         }, e -> {
            container.logger().error(e);

            if (e instanceof NetSocketException) {
               sendResponse(Proto.Msg.newBuilder().setError(e.getMessage()).build(), ((NetSocketException) e).getSocket());
            }
         });

      netServer.coreServer().listen(5555);
      httpServer.coreHttpServer().listen(5556);
      container.logger().info("Started TCP listener at port 5555");
   }

   private void sendResponse(Proto.Msg msg, WriteStream<?> sock) {
      byte[] bytes = msg.toByteArray();
      Buffer response = new Buffer();
      response.appendInt(bytes.length);
      response.appendBytes(bytes);
      sock.write(response);
   }
}

@@ -125,24 +125,76 @@
 HttpServer;%0A
+import io.vertx.rxcore.java.http.RxServerWebSocket;%0A
 import io.ve
@@ -539,28 +539,88 @@
 le;%0A
-%0Aimport java.net.URI
+import rx.Subscription;%0A%0Aimport java.net.URI;%0Aimport java.net.URISyntaxException
 ;%0Aim
@@ -909,21 +909,17 @@
 ocket().
-flatM
+m
 ap(s -%3E 
@@ -1136,25 +1136,24 @@
 ny().get();%0A
-%0A
             
@@ -1163,32 +1163,152 @@
 urn 
-Riemann.getEvents(vertx)
+Tup2.create(s, Query.parse(nameValuePair.getValue()));%0A         %7D catch (URISyntaxException e) %7B%0A            throw new NetSocketException(s, e);
 %0A   
@@ -1317,64 +1317,181 @@
     
+%7D%0A
       
-.filter(Query.parse(nameValuePair.getValue()))%0A   
+%7D).subscribe(r -%3E %7B%0A         RxServerWebSocket socket = r.getLeft();%0A         Subscription subscription = Riemann.getEvents(vertx)%0A            .filter(r.getRight())%0A
     
@@ -1525,19 +1525,16 @@
 bject()%0A
-   
         
@@ -1603,35 +1603,32 @@
 %0A               
-  
- 
 .putString(%22host
@@ -1639,27 +1639,24 @@
 .getHost())%0A
-   
             
@@ -1699,35 +1699,32 @@
 %0A               
- 
-  
 .putString(%22serv
@@ -1752,35 +1752,32 @@
 %0A               
-  
- 
 .putString(%22desc
@@ -1813,35 +1813,32 @@
 %0A               
-   
 .putNumber(%22metr
@@ -1864,35 +1864,32 @@
 %0A               
- 
-  
 .putNumber(%22time
@@ -1911,35 +1911,32 @@
 %0A               
-  
- 
 .putNumber(%22ttl%22
@@ -1966,207 +1966,235 @@
     
-   .map(j -%3E Tup2.create(s, j));%0A%0A         %7D catch (Exception e) %7B%0A            throw new NetSocketException(s, e);%0A    
+.subscribe(json -%3E socket.writeTextFrame(json.encode()));%0A%0A         socket.closeHandler(h -%3E subscription.unsubscribe());%0A      %7D, e -%3E %7B%0A         container.logger().error(e);%0A%0A
      
-%7D%0A
     
-  %7D).subscribe(m -%3E m.getLeft().writeTextFrame(m.getRight().encode())
+//TODO: Fix proper error handling%0A      %7D
 );%0A%0A

Fix websocket handler so that a query can change without stopping the whole subscription.