IoT notifications

import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import java.io.ByteArrayInputStream;

public class AvroConverter<T> {
  private final ThreadLocal<BinaryDecoder> threadLocalDecoder = new ThreadLocal<>();
  private final DatumReader<T> datumReader;

  /**
   * Creates converter instance for specified class type.
   * @param type - object type for which this converter will be initialized
   */
  public AvroConverter(Class<T> type) {
    this.datumReader = new SpecificDatumReader<>(type);
  }

  /**
  * Converts byte array into Java class passed in constructor.
  * @return Java class
  */
  public T decode(byte[] bytes) {
    ByteArrayInputStream byteIn = new ByteArrayInputStream(bytes);
    Decoder decoder = getDecoder(byteIn);
    try {
      return datumReader.read(null, decoder);
    } catch (Exception ex) {
      String msg = "Failed to deserialize submitted byte array into object.";
      throw new RuntimeException(msg, ex);
    }
  }

  private BinaryDecoder getDecoder(ByteArrayInputStream byteIn) {
    BinaryDecoder reuse = threadLocalDecoder.get();
    BinaryDecoder decoder = DecoderFactory.get().directBinaryDecoder(byteIn, reuse);
    if (reuse == null) {
      threadLocalDecoder.set(decoder);
    }
    return decoder;
  }
}

import io.nats.client.Connection;
import io.nats.client.Dispatcher;
import io.nats.client.Message;
import io.nats.client.Nats;
import org.kaaproject.ipc.tstp.gen.v1.TimeSeriesEvent;
import java.util.function.Consumer;

public class TimeSeriesEventService {

  // Inbound message <a href="https://en.wikipedia.org/wiki/Apache_Avro">Avro</a> converter.
  private static final AvroConverter<TimeSeriesEvent> AVRO_CONVERTER = new AvroConverter<>(TimeSeriesEvent.class);

  private static final String NATS_URL = "nats://localhost:4222";

  // Time series and <a href="https://docs.kaaiot.io/EPTS/docs/current/Overview/">EPTS</a> information.
  private static final String TIME_SERIES_NAME = "notification";
  private static final String EPTS_SERVICE_INSTANCE_NAME = "epts";
  private static final String TSTP_SUBSCRIBE_SUBJECT = String
      .format("kaa.v1.events.%s.endpoint.data-collection.data-points-received.%s", EPTS_SERVICE_INSTANCE_NAME, TIME_SERIES_NAME);

  private final Connection natsConnection;

  public TimeSeriesEventService() throws Exception {
    this.natsConnection = Nats.connect(NATS_URL);
  }

  public void onTimeSeriesEvent() {
    Dispatcher dispatcher = natsConnection.createDispatcher((msg) -> handleNatsMessage(msg));
    dispatcher.subscribe(TSTP_SUBSCRIBE_SUBJECT);
    System.out.println(String.format("Subscribed to the time series [%s] events", TIME_SERIES_NAME));
    while (true) {
      // loop forever
    }
  }

  private void handleNatsMessage(Message msg) {
    try {
      TimeSeriesEvent timeSeriesEvent = AVRO_CONVERTER.decode(msg.getData());
      System.out.println(String.format("Handling incoming time series event: %s", timeSeriesEvent));
    } catch (Exception ex) {
      System.err.println(String.format("Exception occurred during handling time series event: %s", ex));
    }
  }
}

import org.kaaproject.ipc.tstp.gen.v1.TimeSeriesEvent;
import java.util.stream.Collectors;

public class NotificationService {

  public static void main(String[] args) throws Exception {
    System.out.println("Starting notification application...");
    TimeSeriesEventService timeSeriesEventService = new TimeSeriesEventService();
    timeSeriesEventService.onTimeSeriesEvent();
  }
}

import org.springframework.mail.SimpleMailMessage;
import org.springframework.mail.javamail.JavaMailSenderImpl;
import java.util.Properties;

public class EmailService {

  private final String USERNAME = "my_username@gmail.com";  // Email from the Gmail account from which emails will be sent.
  private final String PASSWORD = "my_secret_pass";         // Password from the Gmail account.

  private final JavaMailSenderImpl mailSender;

  public EmailService() {
    JavaMailSenderImpl mailSender = new JavaMailSenderImpl();
    mailSender.setHost("smtp.gmail.com");
    mailSender.setPort(587);
    mailSender.setUsername(USERNAME);
    mailSender.setPassword(PASSWORD);

    Properties props = mailSender.getJavaMailProperties();
    props.put("mail.transport.protocol", "smtp");
    props.put("mail.smtp.auth", "true");
    props.put("mail.smtp.starttls.enable", "true");
    props.put("mail.debug", "true");
    this.mailSender = mailSender;
  }

  public void sendNotification(String to, String subject, String text) {
    SimpleMailMessage message = new SimpleMailMessage();
    message.setTo(to);
    message.setSubject(subject);
    message.setText(text);
    this.mailSender.send(message);
  }
}

  public void onTimeSeriesEvent(Consumer<TimeSeriesEvent> onTimeSeriesEventCallback) {
    Dispatcher dispatcher = natsConnection.createDispatcher((msg) -> handleNatsMessage(msg, onTimeSeriesEventCallback));
    dispatcher.subscribe(TSTP_SUBSCRIBE_SUBJECT);
    System.out.println(String.format("Subscribed to the time series [%s] events", TIME_SERIES_NAME));
    while (true) {
      // loop forever
    }
  }

  private void handleNatsMessage(Message msg, Consumer<TimeSeriesEvent> onTimeSeriesEventCallback) {
    try {
      TimeSeriesEvent timeSeriesEvent = AVRO_CONVERTER.decode(msg.getData());
      System.out.println(String.format("Handling incoming time series event: %s", timeSeriesEvent));
      // Calling the event consumer
      onTimeSeriesEventCallback.accept(timeSeriesEvent);
    } catch (Exception ex) {
      System.err.println(String.format("Exception occurred during handling time series event: %s", ex));
    }
  }

  private static String RECIPIENT_EMAIL = "email@example.com"; // Specify recipient of email notifications.

  public static void main(String[] args) throws Exception {
    System.out.println("Starting notification application...");
    EmailService emailService = new EmailService();
    TimeSeriesEventService timeSeriesEventService = new TimeSeriesEventService();
    timeSeriesEventService.onTimeSeriesEvent(timeSeriesEvent -> handleTimeSeriesEvent(timeSeriesEvent, emailService));
  }

  private static void handleTimeSeriesEvent(TimeSeriesEvent event, EmailService emailService) {
    String eventType = event.getDataPoints().stream()
        .map(dataPoint -> dataPoint.getValues().get("temperature").toString())
        .collect(Collectors.joining(", "));
    String emailMessageText = String.format("Next events has occurred: %s", eventType);
    emailService.sendNotification(RECIPIENT_EMAIL, "Smart house notifications", emailMessageText);
  }