change gRPC consume for rest consume

src/main/java/com/cardiff/pocmeasurekafka/in/stream/MeasureConsumer.java

@@ -19,7 +19,7 @@ public class MeasureConsumer implements Consumer<Measure> {
 
   @Override
   public void accept(Measure measure) {
-    measureService.saveMeasureRedis(measure);
+    measureService.saveMeasure(measure);
     log.info("Catch event {}", measure);
   }
 }

src/main/java/com/cardiff/pocmeasurekafka/out/grpc/GrpcAdapter.java

@@ -16,6 +16,7 @@ import com.paloit.pocmeasuregrpc.MeasureServiceGrpc;
 import com.paloit.pocmeasuregrpc.MeasureServiceGrpc.MeasureServiceBlockingStub;
 
 @Component
+@Deprecated(since = "01/08/2023")
 public class GrpcAdapter {
 
   @Value("${grpc.adapter.host:localhost}")
@@ -23,7 +24,7 @@ public class GrpcAdapter {
   @Value("${grpc.adapter.port:9099}")
   private int port;
 
-  public String createMeasure(Measure measure) {
+  /*public String createMeasure(Measure measure) {
     ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port)
         .usePlaintext()
         .build();
@@ -35,5 +36,5 @@ public class GrpcAdapter {
             .build());
     channel.isShutdown();
     return measureResponse.getDeviceId();
-  }
+  }*/
 }

src/main/java/com/cardiff/pocmeasurekafka/out/rest/RestAdapter.java

+package com.cardiff.pocmeasurekafka.out.rest;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+import org.springframework.web.util.UriComponentsBuilder;
+
+import com.cardiff.pocmeasurekafka.model.Measure;
+
+
+@Component
+public class RestAdapter {
+
+  private static final Logger log = LoggerFactory.getLogger(RestAdapter.class);
+  @Value("${m3.api.url:http://localhost:8080/measure}")
+  private String m3ApiUrl;
+
+  public String createMeasure(Measure measure){
+    RestTemplate restTemplate = new RestTemplate();
+    HttpHeaders headers = new HttpHeaders();
+    headers.setContentType(MediaType.APPLICATION_JSON);
+    HttpEntity<MeasureRest> requestEntity = new HttpEntity<>(this.mapper(measure), headers);
+
+    String absoluteUri = UriComponentsBuilder
+        .fromHttpUrl(m3ApiUrl)
+        .toUriString();
+
+    ResponseEntity<String> response = restTemplate.exchange(absoluteUri, HttpMethod.POST, requestEntity, String.class);
+    return response.getBody();
+  }
+
+  private MeasureRest mapper(Measure measure){
+    return new MeasureRest(measure.deviceId(), measure.value(), measure.date().toInstant(ZoneOffset.UTC));
+  }
+
+  public record MeasureRest(String deviceId, String value, Instant date) implements Serializable {
+
+  }
+
+}

src/main/java/com/cardiff/pocmeasurekafka/service/MeasureService.java

@@ -5,8 +5,8 @@ import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 import com.cardiff.pocmeasurekafka.model.Measure;
-import com.cardiff.pocmeasurekafka.out.grpc.GrpcAdapter;
 import com.cardiff.pocmeasurekafka.out.redis.RedisAdapter;
+import com.cardiff.pocmeasurekafka.out.rest.RestAdapter;
 
 import lombok.RequiredArgsConstructor;
 
@@ -16,15 +16,14 @@ public class MeasureService {
 
   private static final Logger log = LoggerFactory.getLogger(MeasureService.class);
   private final RedisAdapter redisAdapter;
-  private final GrpcAdapter grpcAdapter;
-  public String saveMeasureRedis(Measure measure){
+  private final RestAdapter restAdapter;
+  public void saveMeasure(Measure measure){
     String deviceIdRedis = redisAdapter.saveMeasureRedis(measure).deviceId();
-    log.info("catch deviceID response redis {}", deviceIdRedis);
+    log.info("catch deviceID save in redis {}", deviceIdRedis);
     String deviceIdFromRedis = redisAdapter.getByDeviceId(measure.deviceId()).deviceId();
     log.info("catch deviceID getByDeviceId from redis {}", deviceIdFromRedis);
-    String deviceIdGrpc = grpcAdapter.createMeasure(measure);
-    log.info("catch deviceID response Grpc {}", deviceIdRedis);
-    return deviceIdGrpc;
+    String deviceIdFromRest = restAdapter.createMeasure(measure);
+    log.info("catch deviceID from m3 rest service consume {}", deviceIdFromRest);
   }
 
 }

src/main/resources/application-local.properties

@@ -51,3 +51,6 @@ spring.cache.redis.time-to-live=1m
 #GrpcEnpoint
 grpc.adapter.host = localhost
 grpc.adapter.port = 9099
+
+#RestEndpoint
+m3.api.url = http://localhost:9099/measure
\ No newline at end of file