2 files changed, 46 insertions, 64 deletions

diff --git a/juick-api/src/main/java/com/juick/api/TGBot.java b/juick-api/src/main/java/com/juick/api/TGBot.java
index bbfed259..7e261540 100644
--- a/juick-api/src/main/java/com/juick/api/TGBot.java
+++ b/juick-api/src/main/java/com/juick/api/TGBot.java
@@ -4,10 +4,6 @@ import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.juick.service.MessagesService;
 import com.juick.service.TelegramService;
-import com.neovisionaries.ws.client.WebSocket;
-import com.neovisionaries.ws.client.WebSocketAdapter;
-import com.neovisionaries.ws.client.WebSocketFactory;
-import com.neovisionaries.ws.client.WebSocketFrame;
 import com.pengrad.telegrambot.Callback;
 import com.pengrad.telegrambot.TelegramBot;
 import com.pengrad.telegrambot.TelegramBotAdapter;
@@ -19,13 +15,13 @@ import com.pengrad.telegrambot.response.SendResponse;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketSession;
+import org.springframework.web.socket.handler.TextWebSocketHandler;
 
 import javax.inject.Inject;
 import java.io.IOException;
 import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 import static com.juick.formatters.PlainTextFormatter.formatPost;
 import static com.juick.formatters.PlainTextFormatter.formatUrl;
@@ -33,11 +29,10 @@ import static com.juick.formatters.PlainTextFormatter.formatUrl;
 /**
  * Created by vt on 12/05/16.
  */
-public class TGBot implements AutoCloseable {
+public class TGBot extends TextWebSocketHandler {
     private static final Logger logger = LoggerFactory.getLogger(TGBot.class);
 
     TelegramBot bot;
-    WebSocket ws;
     private ObjectMapper ms;
 
     @Inject
@@ -45,8 +40,6 @@ public class TGBot implements AutoCloseable {
     @Inject
     MessagesService messagesService;
 
-    ExecutorService service = Executors.newSingleThreadExecutor();
-
     private TGBot() {
         throw new IllegalStateException();
     }
@@ -65,53 +58,37 @@ public class TGBot implements AutoCloseable {
             if (!bot.execute(webhook).isOk()) {
                 logger.error("error setting webhook");
             }
-            ws = new WebSocketFactory().createSocket("wss://ws.juick.com/");
-            ws.setPingInterval(60 * 1000);
-            ws.addListener(new WebSocketAdapter() {
-                @Override
-                public void onDisconnected(WebSocket websocket, WebSocketFrame serverCloseFrame, WebSocketFrame clientCloseFrame, boolean closedByServer) throws Exception {
-                    logger.info("ws disconnected");
-                    ws.recreate();
-                }
-
-                @Override
-                public void onConnected(WebSocket websocket, Map<String, List<String>> headers) {
-                    logger.info("ws connected");
-                }
-
-                @Override
-                public void onTextMessage(WebSocket websocket, String text) throws Exception {
-                    super.onTextMessage(websocket, text);
-                    com.juick.Message jmsg = ms.readValue(text, com.juick.Message.class);
-
-                    if (logger.isInfoEnabled()) // prevent writeValueAsString execution if logger disabled
-                        logger.info("got jmsg: {}", ms.writeValueAsString(jmsg));
-
-                    String msgUrl = formatUrl(jmsg);
-                    if (jmsg.getRid() == 0) {
-                        String msg = formatPost(jmsg);
-
-                        List<Long> users = telegramService.getSubscribers(jmsg.getUser().getUid());
-                        List<Long> chats = telegramService.getChats();
-                        // registered subscribed users
-
-                        users.forEach(c -> telegramNotify(c, msg, msgUrl));
-                        // anonymous
-                        chats.stream().filter(u -> telegramService.getUser(u) == 0).forEach(c -> telegramNotify(c, msg, msgUrl));
-                    } else {
-                        // get quote
-                        com.juick.Message msg = messagesService.getReply(jmsg.getMid(), jmsg.getRid());
-                        String fmsg = formatPost(msg);
-                        telegramService.getSubscribersToComments(jmsg.getMid(), jmsg.getUser().getUid()).forEach(c -> telegramNotify(c, fmsg, msgUrl));
-                    }
-                }
-            });
-            service.submit(() -> ws.connect());
         } catch (Exception e) {
             logger.warn("couldn't initialize telegram bot", e);
         }
     }
 
+    @Override
+    protected void handleTextMessage(WebSocketSession session, TextMessage text) throws Exception {
+        com.juick.Message jmsg = ms.readValue(text.asBytes(), com.juick.Message.class);
+
+        if (logger.isInfoEnabled()) // prevent writeValueAsString execution if logger disabled
+            logger.info("got jmsg: {}", ms.writeValueAsString(jmsg));
+
+        String msgUrl = formatUrl(jmsg);
+        if (jmsg.getRid() == 0) {
+            String msg = formatPost(jmsg);
+
+            List<Long> users = telegramService.getSubscribers(jmsg.getUser().getUid());
+            List<Long> chats = telegramService.getChats();
+            // registered subscribed users
+
+            users.forEach(c -> telegramNotify(c, msg, msgUrl));
+            // anonymous
+            chats.stream().filter(u -> telegramService.getUser(u) == 0).forEach(c -> telegramNotify(c, msg, msgUrl));
+        } else {
+            // get quote
+            com.juick.Message msg = messagesService.getReply(jmsg.getMid(), jmsg.getRid());
+            String fmsg = formatPost(msg);
+            telegramService.getSubscribersToComments(jmsg.getMid(), jmsg.getUser().getUid()).forEach(c -> telegramNotify(c, fmsg, msgUrl));
+        }
+    }
+
     public void telegramNotify(Long c, String msg, String msgUrl) {
         SendMessage telegramMessage = new SendMessage(c, msg);
         if (msgUrl != null) {
@@ -156,15 +133,4 @@ public class TGBot implements AutoCloseable {
             }
         });
     }
-
-    @Override
-    public void close() throws Exception {
-        try {
-            if (ws != null) {
-                ws.disconnect();
-            }
-        } catch (Exception e) {
-            logger.warn("websocket disconnection exception", e);
-        }
-    }
 }
diff --git a/juick-api/src/main/java/com/juick/api/configuration/ApiAppConfiguration.java b/juick-api/src/main/java/com/juick/api/configuration/ApiAppConfiguration.java
index 409734e4..1781f5a6 100644
--- a/juick-api/src/main/java/com/juick/api/configuration/ApiAppConfiguration.java
+++ b/juick-api/src/main/java/com/juick/api/configuration/ApiAppConfiguration.java
@@ -7,6 +7,8 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.PropertySource;
 import org.springframework.core.env.Environment;
+import org.springframework.web.socket.client.WebSocketConnectionManager;
+import org.springframework.web.socket.client.standard.StandardWebSocketClient;
 
 import javax.inject.Inject;
 
@@ -28,4 +30,18 @@ public class ApiAppConfiguration {
     public TGBot tgBot() {
         return new TGBot(env.getProperty("telegram_token", StringUtils.EMPTY));
     }
+
+    private static final String WS_URI = "wss://ws.juick.com/";
+
+    @Bean
+    public WebSocketConnectionManager connectionManager() {
+        WebSocketConnectionManager manager = new WebSocketConnectionManager(client(), tgBot(), WS_URI);
+        manager.setAutoStartup(true);
+        return manager;
+    }
+
+    @Bean
+    public StandardWebSocketClient client() {
+        return new StandardWebSocketClient();
+    }
 }