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?
