2
\$\begingroup\$

I implemented observer pattern and it will be great if someone can do review of my work and say what should be done better.

What is the problem. Application monitoring rss channel, if there is something new then should do 3 things: send email to registered user that following stock which has news (only EBI and ESPI channels) send any news message to chat (here im using web sockets) and save the news message to DB.

Here is my interface:

public interface RssObservable {

    void addRssObserver(RssObserver observer);
    void removeRssObserver(RssObserver observer);
    void notifyRssObservers(List<NewsMessage> parsedRssNewsMessage);
    void fireCron();
}

My implementation:

@Service
public class RssParserService implements RssObservable {

    private final Logger logger = LoggerFactory.getLogger(RssParserService.class);

    @Inject
    private StockRepository stockRepository;

    @Inject
    private UserRepository userRepository;

    private Map<RssType, LocalDateTime> rssLinks;
    private Set<RssObserver> observers;
    private List<NewsMessage> parsedRssNewsMessage;
    private FeedFetcher feedFetcher;

    @PostConstruct
    private void prepareMapWithLinks() {
        rssLinks = new HashMap<>();
        observers = new HashSet<>();
        feedFetcher = new HttpURLFeedFetcher();

        LocalDateTime date = new LocalDateTime();
        rssLinks.put(RssType.CHALLENGE, new LocalDateTime(date));
        rssLinks.put(RssType.EBI, new LocalDateTime(date));
        rssLinks.put(RssType.ESPI, new LocalDateTime(date));
        rssLinks.put(RssType.PAP, new LocalDateTime(date));
        rssLinks.put(RssType.RECOMMENDATIONS, new LocalDateTime(date));
        rssLinks.put(RssType.RESULTS, new LocalDateTime(date));
    }

    @Override
    @Scheduled(cron = "*/5 * 8-17 * * MON-FRI")
    @Scheduled(cron = "0 */5 18-23 * * MON-FRI")
    @Scheduled(cron = "0 */30 0-7 * * MON-FRI")
    @Scheduled(cron = "0 */30 * * * SAT,SUN")
    public void fireCron() {
        parsedRssNewsMessage = new ArrayList<>();
        if (isNewRssMessagePresented()) {
            notifyRssObservers(parsedRssNewsMessage);
        }
    }

    private boolean isNewRssMessagePresented() {
        for (Map.Entry<RssType, LocalDateTime> entry : rssLinks.entrySet()) {
            parsedRssNewsMessage.addAll(getNewsMessagesFromUrl(entry.getKey(), entry.getValue()));
        }
        return !parsedRssNewsMessage.isEmpty();
    }

    private List<NewsMessage> getNewsMessagesFromUrl(RssType rssType, LocalDateTime date) {
        SyndFeed feed = null;
        List<NewsMessage> rssNewsMessages = new ArrayList<>();
        try {
            feed = feedFetcher.retrieveFeed(new URL(rssType.getUrl()));
            List<SyndEntry> syndFeedItems = feed.getEntries();

            int indexOfLatestItem = 0;
            LocalDateTime dt = new LocalDateTime(syndFeedItems.get(indexOfLatestItem).getPublishedDate());

            for (Object syndFeedEntry : syndFeedItems) {
                SyndEntry syndEntry = (SyndEntry) syndFeedEntry;
                LocalDateTime syndEntryPublishDate = new LocalDateTime(syndEntry.getPublishedDate());
                if (syndEntryPublishDate.isAfter(date)) {
                    NewsMessage message = parseNewsMessageFromRssChannel(syndEntry, rssType);
                    rssNewsMessages.add(message);
                } else {
                    break;
                }
                rssLinks.put(rssType, dt);
            }
        } catch (IllegalArgumentException | IOException | FeedException | FetcherException e) {
            logger.error("error occurs", e.getMessage());
        }

        return rssNewsMessages;
    }

    private NewsMessage parseNewsMessageFromRssChannel(SyndEntry syndEntry, RssType type) {
        String message = syndEntry.getTitle();
        String link = syndEntry.getLink();
        ZonedDateTime date = ZonedDateTime.ofInstant(syndEntry.getPublishedDate().toInstant(), ZoneId.systemDefault());
        NewsMessage stNwMsg = new NewsMessage();
        stNwMsg.setType(type);
        stNwMsg.setMessage(message);
        stNwMsg.setLink(link);
        stNwMsg.setCreatedDate(date);
        stNwMsg.setUserLogin("system");
        stNwMsg.setUser(userRepository.findOneByLogin("system").get());
        if (type.equals(RssType.EBI) || type.equals(RssType.ESPI)) {
            if (getStockFromTitle(message).isPresent()) {
                stNwMsg.setStock(getStockFromTitle(message).get());
            }
        }
        return stNwMsg;
    }

    private Optional<Stock> getStockFromTitle(String message) {
        Pattern pattern = Pattern.compile("^([0-9A-Z-/.]+ )+");
        Matcher matcher = pattern.matcher(message);
        if (matcher.find()) {
            Optional<Stock> stock = stockRepository.findByStockName(matcher.group(0).trim());
            if (stock.isPresent()) {
                return Optional.of(stock.get());
            }
        }
        return Optional.empty();
    }

    @Override
    public void addRssObserver(RssObserver observer) {
        observers.add(observer);
    }

    @Override
    public void removeRssObserver(RssObserver observer) {
        observers.remove(observer);
    }

    @Override
    public void notifyRssObservers(List<NewsMessage> parsedRssNewsMessage) {
        observers.forEach(e -> e.updateRssNewsMessage(parsedRssNewsMessage));
    }

}

My observers:

public interface RssObserver {
    void updateRssNewsMessage(List<NewsMessage> parsedRssNewsMessage);
}

Sockets observer:

@Service
public class SocketMessageService implements RssObserver, SocketMessageHandler {

    @Inject
    private SimpMessageSendingOperations messagingTemplate;

    @Inject
    private RssObservable rssParserService;

    @PostConstruct
    private void init() {
        rssParserService.addRssObserver(this);
    }

    public void sendMostActiveStocksToClient(List<StockFiveMinutesDetails> stockFiveMinutesDetails, LocalTime time) {
        TimeStockFiveMinuteDetails timeStockFiveMinuteDetails = new TimeStockFiveMinuteDetails();
        timeStockFiveMinuteDetails.setTime(time);
        timeStockFiveMinuteDetails.setListOfDetails(stockFiveMinutesDetails);
        messagingTemplate.convertAndSend("/most/active/stocks", timeStockFiveMinuteDetails);
    }

    public void sendToChat(NewsMessage message) {
        messagingTemplate.convertAndSend("/webchat/recive", (ChatMessage) message);
    }

    @Override
    public void updateRssNewsMessage(List<NewsMessage> parsedRssNewsMessage) {
        parsedRssNewsMessage.stream()
            .sorted((e1, e2) -> e1.getCreatedDate().compareTo(e2.getCreatedDate()))
            .forEach(e -> sendToChat(e));
    }
}

Database news observer:

@Service
public class ChatMessageService implements RssObserver, ChatMessageLoader {

    @Inject
    private ChatMessageRepository chatMessageRepository;

    @Inject
    private RssObservable rssParserService;

    @PostConstruct
    private void init() {
        rssParserService.addRssObserver(this);
    }

    public ResponseEntity<List<ChatMessage>> getLastMessages(int page) {
        List<ChatMessage> reverse = getMessages(page);
        Collections.reverse(reverse);
        return new ResponseEntity<List<ChatMessage>>(reverse, HttpStatus.OK);
    }

    public ResponseEntity<List<ChatMessage>> getOlderMessages(int page) {
        List<ChatMessage> messages = getMessages(page);
        return new ResponseEntity<List<ChatMessage>>(messages, HttpStatus.OK);
    }

    private List<ChatMessage> getMessages(int page) {
        Pageable pageRequest = new PageRequest(page, 10, Sort.Direction.DESC, "createdDate");
        Page<ChatMessage> messages = chatMessageRepository.findAll(pageRequest);
        List<ChatMessage> reverse = new ArrayList<ChatMessage>(messages.getContent());
        return reverse;
    }

    @Override
    public void updateRssNewsMessage(List<NewsMessage> parsedRssNewsMessage) {
        chatMessageRepository.save(parsedRssNewsMessage);
    }
}

And mail observer:

@Service
public class MailService implements RssObserver, MailSender {

    private final Logger log = LoggerFactory.getLogger(MailService.class);

    (...)

    @PostConstruct
    private void init() {
        rssParserService.addRssObserver(this);
    }
(...)

    @Async
    public void sendBccEmail(String[] to, String subject, String content, boolean isMultipart, boolean isHtml) {

        // Prepare message using a Spring helper
        MimeMessage mimeMessage = javaMailSender.createMimeMessage();
        try {
            MimeMessageHelper message = new MimeMessageHelper(mimeMessage, isMultipart, CharEncoding.UTF_8);
            message.setBcc(to);
            message.setFrom(jHipsterProperties.getMail().getFrom());
            message.setSubject(subject);
            message.setText(content, isHtml);
            javaMailSender.send(mimeMessage);
            log.debug("Sent e-mail to User '{}'", to);
        } catch (Exception e) {
            logger.error("error occurs", e.getMessage());
        }
    }

    public void informUserAboutStockNewsByEmail(NewsMessage message) {
        if (message.getStock() == null) {
            return;
        }
        List<User> usersToSendEmail = userRepository.findAllByStocks(message.getStock());
        String[] emails = usersToSendEmail.stream().map(e -> e.getEmail()).toArray(size -> new String[size]);
        String mailTopic = "[Kanal: " + message.getType().toString() + "] [" + message.getStock().getTicker().toString().toUpperCase() + "]" + message.getMessage();

        sendBccEmail(emails, mailTopic, message.getChatMessage(), false, true);
    }

    @Override
    public void updateRssNewsMessage(List<NewsMessage> parsedRssNewsMessage) {
        parsedRssNewsMessage.forEach(e -> informUserAboutStockNewsByEmail(e));
    }
}

And my question regards to observer pattern: Is it necessary to add observers to list in my way? Or can I do it like this: Inject this three services to my RssParserService and in my PostConstruct method add it to the list of observers?

\$\endgroup\$
3
\$\begingroup\$

RssParserService should NOT know concretely about its observers. It would be nearly the same as letting an abstract class know its derived classes. All your decoupling benefits through abstraction/dependency inversion (here the observer pattern) would be gone.

On the other side your own observer implementations relies now on a dependency injection mechanism (@Inject) AND service construction (@PostConstruct). This looks really elegant at the first place. But me, I wouldn't make MY observer implementation depend on OTHER mechanisms.

What you are missing is a model where your services depend on. In the use case presented you will never have any real problems. But if your code grows you should ...

  1. ... separate the technologies (Spring service, Spring DI, Spring task scheduling) ...
  2. ... introduce a model ...
  3. ... make the services depend on the model (observer mechanism) and not on each other ...

... to have a technology free core.

Finally I only see the method "getNewsMessagesFromUrl(..)" to have potential to be taken apart. Everything else seems to me pretty straight forward.

\$\endgroup\$
  • \$\begingroup\$ Awesome, thank you very much for your review. Regards \$\endgroup\$ – Pulkownik Dec 28 '15 at 9:35

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.