diff options
Diffstat (limited to 'src/main/java/com/juick/server')
4 files changed, 3 insertions, 317 deletions
diff --git a/src/main/java/com/juick/server/ServerManager.java b/src/main/java/com/juick/server/ServerManager.java index 7a95ec43..be537409 100644 --- a/src/main/java/com/juick/server/ServerManager.java +++ b/src/main/java/com/juick/server/ServerManager.java @@ -16,7 +16,6 @@ */ package com.juick.server; -import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.juick.Message; import com.juick.User; @@ -24,13 +23,7 @@ import com.juick.model.AnonymousUser; import com.juick.service.MessagesService; import com.juick.service.SubscriptionService; import com.juick.service.UserService; -import com.juick.service.component.LikeEvent; -import com.juick.service.component.MessageEvent; -import com.juick.service.component.MessageReadEvent; -import com.juick.service.component.NotificationListener; -import com.juick.service.component.PingEvent; -import com.juick.service.component.SubscribeEvent; -import com.juick.service.component.TopEvent; +import com.juick.service.component.*; import com.juick.util.MessageUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -38,11 +31,9 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; -import org.springframework.web.socket.TextMessage; import javax.annotation.PostConstruct; import javax.inject.Inject; -import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -62,8 +53,6 @@ public class ServerManager implements NotificationListener { @Inject private MessagesService messagesService; @Inject - private WebsocketManager wsHandler; - @Inject private SubscriptionService subscriptionService; @Inject private UserService userService; @@ -81,96 +70,15 @@ public class ServerManager implements NotificationListener { private void onJuickPM(final User to, final com.juick.Message jmsg) { - try { - String json = jsonMapper.writeValueAsString(jmsg); - synchronized (wsHandler.getClients()) { - wsHandler.getClients().stream().filter(c -> - (!c.legacy && c.visitor.getUid() == to.getUid()) || c.visitor.equals(serviceUser)) - .forEach(c -> { - try { - logger.debug("sending pm to {}", c.visitor.getUid()); - c.sendMessage(new TextMessage(json)); - } catch (IOException e) { - logger.warn("ws error", e); - } - }); - } - } catch (JsonProcessingException e) { - logger.warn("Invalid JSON", e); - } messageEvent(jmsg, Collections.singletonList(to)); } private void onJuickMessagePost(final com.juick.Message jmsg, List<User> subscribedUsers) { - try { - String json = jsonMapper.writeValueAsString(jmsg); - List<Integer> uids = subscribedUsers - .stream().map(User::getUid).collect(Collectors.toList()); - synchronized (wsHandler.getClients()) { - wsHandler.getClients().stream().filter(c -> - (!c.legacy && c.visitor.isAnonymous()) // anonymous users - || c.visitor.equals(serviceUser) // services - || (!c.legacy && uids.contains(c.visitor.getUid()))) // subscriptions - .forEach(c -> { - try { - logger.debug("sending message to {}", c.visitor.getUid()); - c.sendMessage(new TextMessage(json)); - } catch (IOException e) { - logger.warn("ws error", e); - } - }); - wsHandler.getClients().stream().filter(c -> - c.legacy && c.allMessages) // legacy all posts - .forEach(c -> { - try { - logger.debug("sending message to legacy client {}", c.visitor.getUid()); - c.sendMessage(new TextMessage(json)); - } catch (IOException e) { - logger.warn("ws error", e); - } - }); - } - } catch (JsonProcessingException e) { - logger.warn("Invalid JSON", e); - } messageEvent(jmsg, subscribedUsers); messageEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE)); } private void onJuickMessageReply(final com.juick.Message jmsg, final List<User> subscribedUsers) { - try { - - String json = jsonMapper.writeValueAsString(jmsg); - List<Integer> threadUsers = - subscribedUsers - .stream().map(User::getUid).collect(Collectors.toList()); - synchronized (wsHandler.getClients()) { - wsHandler.getClients().stream().filter(c -> - (!c.legacy && c.visitor.isAnonymous()) // anonymous users - || c.visitor.equals(serviceUser) // services - || (!c.legacy && threadUsers.contains(c.visitor.getUid()))) // subscriptions - .forEach(c -> { - try { - logger.debug("sending reply to {}", c.visitor.getUid()); - c.sendMessage(new TextMessage(json)); - } catch (IOException e) { - logger.warn("ws error", e); - } - }); - wsHandler.getClients().stream().filter(c -> - (c.legacy && c.allReplies) || (c.legacy && c.MID == jmsg.getMid())) // legacy replies - .forEach(c -> { - try { - logger.debug("sending reply to legacy client {}", c.visitor.getUid()); - c.sendMessage(new TextMessage(json)); - } catch (IOException e) { - logger.warn("ws error", e); - } - }); - } - } catch (JsonProcessingException e) { - logger.warn("Invalid JSON", e); - } messageEvent(jmsg, subscribedUsers); messageEvent(jmsg, Collections.singletonList(AnonymousUser.INSTANCE)); } @@ -226,15 +134,6 @@ public class ServerManager implements NotificationListener { serviceMessage.setUser(user); serviceMessage.setMid(source.getMid()); serviceMessage.setUnread(false); - wsHandler.getClients().stream().filter(c -> - (!c.legacy && c.visitor == user) || c.visitor.equals(serviceUser) - ).forEach(u -> { - try { - u.sendMessage(new TextMessage(jsonMapper.writeValueAsString(serviceMessage))); - } catch (IOException e) { - logger.error("JSON error", e); - } - }); readEvent(serviceMessage, Collections.singletonList(serviceUser)); } diff --git a/src/main/java/com/juick/server/WebsocketManager.java b/src/main/java/com/juick/server/WebsocketManager.java deleted file mode 100644 index 1b62b984..00000000 --- a/src/main/java/com/juick/server/WebsocketManager.java +++ /dev/null @@ -1,174 +0,0 @@ -/* - * Copyright (C) 2008-2017, Juick - * - * This program is free software: you can redistribute it and/or modify - * it under the terms of the GNU Affero General Public License as - * published by the Free Software Foundation, either version 3 of the - * License, or (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Affero General Public License for more details. - * - * You should have received a copy of the GNU Affero General Public License - * along with this program. If not, see <http://www.gnu.org/licenses/>. - */ - -package com.juick.server; - -import com.juick.User; -import com.juick.model.AnonymousUser; -import com.juick.model.CommandResult; -import com.juick.server.util.HttpForbiddenException; -import com.juick.server.util.HttpNotFoundException; -import com.juick.service.MessagesService; -import com.juick.service.UserService; -import org.apache.commons.lang3.StringUtils; -import org.apache.commons.lang3.math.NumberUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.springframework.http.HttpHeaders; -import org.springframework.scheduling.annotation.Scheduled; -import org.springframework.stereotype.Component; -import org.springframework.web.socket.CloseStatus; -import org.springframework.web.socket.PingMessage; -import org.springframework.web.socket.TextMessage; -import org.springframework.web.socket.WebSocketSession; -import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator; -import org.springframework.web.socket.handler.TextWebSocketHandler; -import org.springframework.web.util.UriComponents; -import org.springframework.web.util.UriComponentsBuilder; - -import javax.annotation.Nonnull; -import javax.inject.Inject; -import java.io.IOException; -import java.net.URI; -import java.time.Instant; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; - -/** - * Created by vitalyster on 28.06.2016. - */ -@Component -public class WebsocketManager extends TextWebSocketHandler { - private static final Logger logger = LoggerFactory.getLogger(WebsocketManager.class); - - private final List<UserSession> clients = new CopyOnWriteArrayList<>(); - - @Inject - private UserService userService; - @Inject - private MessagesService messagesService; - @Inject - private CommandsManager commandsManager; - - - @Override - public void afterConnectionEstablished(WebSocketSession session) { - - UserSession userSession = new UserSession(session); - URI hLocation = session.getUri(); - - // Auth - UriComponents uriComponents = UriComponentsBuilder.fromUri(hLocation).build(); - List<String> hash = uriComponents.getQueryParams().get("hash"); - if (hash != null && hash.get(0).length() == 16) { - userSession.visitor = userService.getUserByHash(hash.get(0)); - } else { - logger.debug("wrong hash for {} from {}", userSession.visitor.getUid(), userSession); - } - - if (hLocation.getPath().equals("/ws/")) { - logger.debug("user {} connected", userSession.visitor.getUid()); - } else if (hLocation.getPath().equals("/ws/_all")) { - logger.debug("user {} connected to legacy _all ({})", userSession.visitor.getUid(), hLocation.getPath()); - userSession.legacy = true; - userSession.allMessages = true; - } else if (hLocation.getPath().equals("/ws/_replies")) { - logger.debug("user {} connected to legacy _replies ({})", userSession.visitor.getUid(), hLocation.getPath()); - userSession.legacy = true; - userSession.allReplies = true; - } else if (hLocation.getPath().matches("^/ws/(\\d)+$")) { - int MID = NumberUtils.toInt(hLocation.getPath().substring(4), 0); - if (MID > 0) { - if (messagesService.canViewThread(MID, userSession.visitor.getUid())) { - logger.debug("user {} connected to legacy thread ({}) from {}", userSession.visitor.getUid(), MID, userSession); - userSession.legacy = true; - userSession.MID = MID; - } else { - throw new HttpForbiddenException(); - } - } - } else { - throw new HttpNotFoundException(); - } - clients.add(userSession); - logger.debug("{} clients connected", clients.size()); - } - - @Override - public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { - logger.debug("session closed with status {}: {}", status.getCode(), status.getReason()); - clients.removeIf(c -> c.getDelegate().getId().equals(session.getId())); - logger.debug("{} clients connected", clients.size()); - } - - @Scheduled(fixedRate = 30000) - public void ping() { - clients.forEach(c -> { - try { - if (c.isOpen()) { - c.sendMessage(new PingMessage()); - } - } catch (IOException e) { - logger.error("WebSocket PING exception", e); - } - }); - } - - @Override - protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { - UserSession ws = clients.stream().filter(c -> c.getDelegate().equals(session)) - .findFirst().orElseThrow(IllegalStateException::new); - if (!ws.visitor.isAnonymous()) { - String command = message.getPayload().trim(); - if (StringUtils.isNotEmpty(command)) { - CommandResult result = commandsManager.processCommand(ws.visitor, command, URI.create("")); - ws.sendMessage(new TextMessage(result.getText())); - } - } else { - ws.sendMessage(new TextMessage("Authorization required")); - } - } - - public List<UserSession> getClients() { - return clients; - } - - class UserSession extends ConcurrentWebSocketSessionDecorator { - User visitor; - int MID; - boolean allMessages; - boolean allReplies; - Instant tsConnected; - Instant tsLastData; - boolean legacy; - - UserSession(WebSocketSession session) { - super(session, 60000, 65536); - this.visitor = AnonymousUser.INSTANCE; - tsConnected = tsLastData = Instant.now(); - } - - @Nonnull - @Override - public String toString() { - HttpHeaders headers = getHandshakeHeaders(); - return headers.getOrDefault("X-Real-IP", - Collections.singletonList(getRemoteAddress().toString())).get(0); - } - } -} diff --git a/src/main/java/com/juick/server/api/Index.java b/src/main/java/com/juick/server/api/Index.java index 56f01370..4573641b 100644 --- a/src/main/java/com/juick/server/api/Index.java +++ b/src/main/java/com/juick/server/api/Index.java @@ -17,10 +17,7 @@ package com.juick.server.api; -import com.juick.Status; -import com.juick.server.WebsocketManager; import org.springframework.http.HttpStatus; -import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; @@ -28,7 +25,6 @@ import org.springframework.web.bind.annotation.RestController; import org.springframework.web.servlet.support.ServletUriComponentsBuilder; import springfox.documentation.annotations.ApiIgnore; -import javax.inject.Inject; import java.net.URI; /** @@ -36,19 +32,11 @@ import java.net.URI; */ @RestController public class Index { - @Inject - private WebsocketManager wsHandler; @ApiIgnore - @RequestMapping(value = { "/api/", "/ws/" }, method = RequestMethod.GET, headers = "Connection!=Upgrade") + @RequestMapping(value = { "/api/", "/ws/" }, method = RequestMethod.GET) public ResponseEntity<Void> description() { URI redirectUri = ServletUriComponentsBuilder.fromCurrentRequestUri().path("/swagger-ui.html").build().toUri(); return ResponseEntity.status(HttpStatus.MOVED_PERMANENTLY).location(redirectUri).build(); } - @ApiIgnore - @RequestMapping(value = "/api/status", method = RequestMethod.GET, - produces = MediaType.APPLICATION_JSON_UTF8_VALUE, headers = "Connection!=Upgrade") - public Status status() { - return Status.getStatus(String.valueOf(wsHandler.getClients().size())); - } } diff --git a/src/main/java/com/juick/server/configuration/ApiAppConfiguration.java b/src/main/java/com/juick/server/configuration/ApiAppConfiguration.java index 5a5d2c7b..68b3d35f 100644 --- a/src/main/java/com/juick/server/configuration/ApiAppConfiguration.java +++ b/src/main/java/com/juick/server/configuration/ApiAppConfiguration.java @@ -17,25 +17,15 @@ package com.juick.server.configuration; -import com.juick.server.WebsocketManager; import com.juick.server.api.rss.MessagesView; import com.juick.server.api.rss.RepliesView; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.core.Ordered; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; import org.springframework.web.servlet.view.BeanNameViewResolver; import org.springframework.web.servlet.view.feed.AbstractRssFeedView; -import org.springframework.web.socket.config.annotation.EnableWebSocket; -import org.springframework.web.socket.config.annotation.ServletWebSocketHandlerRegistry; -import org.springframework.web.socket.config.annotation.WebSocketConfigurer; -import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; -import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean; - -import javax.annotation.Nonnull; -import javax.inject.Inject; /** * Created by aalexeev on 11/12/16. @@ -43,24 +33,7 @@ import javax.inject.Inject; @Configuration @EnableAsync(proxyTargetClass = true) @EnableScheduling -@EnableWebSocket -public class ApiAppConfiguration implements WebMvcConfigurer, WebSocketConfigurer { - @Inject - private WebsocketManager websocketManager; - - @Override - public void registerWebSocketHandlers(@Nonnull WebSocketHandlerRegistry registry) { - ((ServletWebSocketHandlerRegistry) registry).setOrder(Ordered.HIGHEST_PRECEDENCE); - registry.addHandler(websocketManager, "/ws/**").setAllowedOrigins("*"); - } - - @Bean - public ServletServerContainerFactoryBean createWebSocketContainer() { - ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); - container.setMaxTextMessageBufferSize(8192); - container.setMaxBinaryMessageBufferSize(8192); - return container; - } +public class ApiAppConfiguration implements WebMvcConfigurer { @Bean public BeanNameViewResolver beanNameViewResolver() { return new BeanNameViewResolver(); |