Add smtp output binding

FineCollectionService/pom.xml

@@ -38,6 +38,11 @@
 			<version>2.8.0</version>
 			<scope>test</scope>
 		</dependency>
+        <dependency>
+            <groupId>org.freemarker</groupId>
+            <artifactId>freemarker</artifactId>
+            <version>2.3.31</version>
+        </dependency>
     </dependencies>
 
     <build>

FineCollectionService/src/main/java/dapr/fines/FineCollectionConfiguration.java

@@ -4,6 +4,10 @@
 import dapr.fines.fines.FineCalculator;
 import dapr.fines.vehicle.DefaultVehicleRegistrationClient;
 import dapr.fines.vehicle.VehicleRegistrationClient;
+import dapr.fines.violation.DaprViolationProcessor;
+import dapr.fines.violation.DefaultViolationProcessor;
+import dapr.fines.violation.ViolationProcessor;
+
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -43,6 +47,17 @@ public RestTemplate restTemplate() {
     public VehicleRegistrationClient vehicleRegistrationClient(final RestTemplate restTemplate) {
         return new DefaultVehicleRegistrationClient(restTemplate, vehicleInformationAddress);
    }
+
+   @Bean
+   public ViolationProcessor violationProcessor(final FineCalculator fineCalculator, final VehicleRegistrationClient vehicleRegistrationClient) {
+       return new DefaultViolationProcessor(fineCalculator, vehicleRegistrationClient);
+   }
+
+
+//    @Bean
+//    public ViolationProcessor violationProcessor(final DaprClient daprClient, final FineCalculator fineCalculator, final VehicleRegistrationClient vehicleRegistrationClient) {
+//        return new DaprViolationProcessor(daprClient, fineCalculator, vehicleRegistrationClient);
+//    }
 
 //    @Bean
 //    public DaprClient daprClient() {

FineCollectionService/src/main/java/dapr/fines/violation/DaprViolationProcessor.java

@@ -0,0 +1,81 @@
+package dapr.fines.violation;
+
+import dapr.fines.fines.FineCalculator;
+import dapr.fines.vehicle.VehicleInfo;
+import dapr.fines.vehicle.VehicleRegistrationClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.StringWriter;
+import java.time.format.DateTimeFormatter;
+import java.util.Map;
+
+import dapr.traffic.violation.SpeedingViolation;
+import freemarker.template.Configuration;
+import freemarker.template.Template;
+import freemarker.template.TemplateExceptionHandler;
+import io.dapr.client.DaprClient;
+
+public class DaprViolationProcessor implements ViolationProcessor {
+    private static final Logger log = LoggerFactory.getLogger(DefaultViolationProcessor.class);
+
+    private static final String BINDING_NAME = "smtp";
+    private static final String BINDING_OPERATION = "create";
+
+    private final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("LLLL, dd y");
+    private final DateTimeFormatter TIME_FORMAT = DateTimeFormatter.ofPattern("HH:mm:ss");
+
+    private final DaprClient daprClient;
+    private final FineCalculator fineCalculator;
+    private final VehicleRegistrationClient vehicleRegistrationClient;
+    private final Configuration templateConfiguration;
+
+    public DaprViolationProcessor(final DaprClient daprClient,
+                                final FineCalculator fineCalculator,
+                                final VehicleRegistrationClient vehicleRegistrationClient) {
+        this.daprClient = daprClient;
+        this.fineCalculator = fineCalculator;
+        this.vehicleRegistrationClient = vehicleRegistrationClient;
+        this.templateConfiguration = new Configuration(Configuration.VERSION_2_3_30);
+        this.templateConfiguration.setClassForTemplateLoading(DefaultViolationProcessor.class, "/email");
+        this.templateConfiguration.setDefaultEncoding("UTF-8");
+        this.templateConfiguration.setTemplateExceptionHandler(TemplateExceptionHandler.RETHROW_HANDLER);
+    }
+
+    public void processSpeedingViolation(final SpeedingViolation violation) {
+        int fine = fineCalculator.calculateFine(violation.excessSpeed());
+        final VehicleInfo vehicleInfo = vehicleRegistrationClient.getVehicleInfo(violation.licenseNumber());
+
+        final String nowDate = DATE_FORMAT.format(violation.timestamp());
+        final String dateSpeedingViolation = DATE_FORMAT.format(violation.timestamp());
+        final String timeSpeedingViolation = TIME_FORMAT.format(violation.timestamp());
+        final Map<String, Object> metadataEmailTemplate = Map.of(
+            "customerName", vehicleInfo.ownerName(),
+            "fineDate", nowDate,
+            "vehicleBrand", vehicleInfo.make(),
+            "vehicleModel", vehicleInfo.model(),
+            "vehicleLicenseNumber", violation.licenseNumber(),
+            "road", violation.roadId(),
+            "timeOfDay", timeSpeedingViolation,
+            "violationDate", dateSpeedingViolation,
+            "excessSpeed", violation.excessSpeed(),
+            "fineAmount", fine);
+        Template template;
+        try {
+            template = templateConfiguration.getTemplate("email-template.ftl");
+            final StringWriter writer = new StringWriter();
+            template.process(metadataEmailTemplate, writer);
+            writer.flush();
+            final String emailBody = writer.toString();
+
+            final Map<String, String> metadataBinding = Map.of(
+            "emailFrom", "donotreply@roadtothe.cloud",
+            "emailTo", "pmalarme@roadtothe.cloud",
+            "subject", "Speeding violation on the " + violation.roadId() + " for vehicle " + violation.licenseNumber(),
+            "priority", "1");
+        daprClient.invokeBinding(BINDING_NAME, BINDING_OPERATION, emailBody, metadataBinding, String.class).block();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+}

FineCollectionService/src/main/java/dapr/fines/violation/DefaultViolationProcessor.java

@@ -0,0 +1,56 @@
+package dapr.fines.violation;
+
+import dapr.fines.fines.FineCalculator;
+import dapr.fines.vehicle.VehicleInfo;
+import dapr.fines.vehicle.VehicleRegistrationClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.format.DateTimeFormatter;
+
+import dapr.traffic.violation.SpeedingViolation;
+
+public class DefaultViolationProcessor implements ViolationProcessor {
+    private static final Logger log = LoggerFactory.getLogger(DefaultViolationProcessor.class);
+
+    private final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("LLLL, dd y");
+    private final DateTimeFormatter TIME_FORMAT = DateTimeFormatter.ofPattern("HH:mm:ss");
+
+    private final FineCalculator fineCalculator;
+    private final VehicleRegistrationClient vehicleRegistrationClient;
+
+    public DefaultViolationProcessor(final FineCalculator fineCalculator,
+                                final VehicleRegistrationClient vehicleRegistrationClient) {
+        this.fineCalculator = fineCalculator;
+        this.vehicleRegistrationClient = vehicleRegistrationClient;
+    }
+
+    public void processSpeedingViolation(final SpeedingViolation violation) {
+        int fine = fineCalculator.calculateFine(violation.excessSpeed());
+        String fineText = fine == -1 ? "to be decided by the prosecutor" : String.format("EUR %.2f", (float) fine);
+        final VehicleInfo vehicleInfo = vehicleRegistrationClient.getVehicleInfo(violation.licenseNumber());
+
+        final String fineMessage = constructLogMessage(violation, vehicleInfo, fineText);
+        log.info(fineMessage);
+    }
+
+    private String constructLogMessage(final SpeedingViolation violation, final VehicleInfo vehicleInfo, final String fineText) {
+        final String date = DATE_FORMAT.format(violation.timestamp());
+        final String time = TIME_FORMAT.format(violation.timestamp());
+
+        return String.format("""
+                        Sent fine notification
+                        \t\t\tTo %s, registered owner of license number %s.
+                        \t\t\tViolation of %d km/h detected on the %s road on %s at %s.
+                        \t\t\tFine: %s.%n
+                        """,
+                vehicleInfo.ownerName(),
+                violation.licenseNumber(),
+                violation.excessSpeed(),
+                violation.roadId(),
+                date,
+                time,
+                fineText
+        );
+    }
+}

FineCollectionService/src/main/java/dapr/fines/violation/KafkaViolationConsumer.java

@@ -14,8 +14,7 @@ public KafkaViolationConsumer(final ViolationProcessor violationProcessor) {
 
 	@KafkaListener(topics = "test", groupId = "test", containerFactory = "kafkaListenerContainerFactory")
     public void listen(SpeedingViolation violation) {
-
-		violationProcessor.processSpeedingViolation(violation);
+        violationProcessor.processSpeedingViolation(violation);
     }