Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Strict synchronous communication to FrontierAPI #16

Open
wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Strict synchronous communication to FrontierAPI #16

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3437dcf
Strict synchronous communication to FrontierAPI
snackk Feb 18, 2021
3e7bbc8
Added maven modules
snackk Feb 23, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
0 annotation-processor/.gitignore → .gitignore
View file
Open in desktop
File renamed without changes.
8 changes: 4 additions & 4 deletions annotation-processor/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.3.RELEASE</version>
<version>2.4.3</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.frontier.api</groupId>
Expand Down Expand Up @@ -42,7 +42,7 @@
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>29.0-jre</version>
<version>30.1-jre</version>
</dependency>

<dependency>
Expand All @@ -60,13 +60,13 @@
<dependency>
<groupId>io.vavr</groupId>
<artifactId>vavr-jackson</artifactId>
<version>0.10.2</version>
<version>0.10.3</version>
</dependency>

<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.12</version>
<version>1.18.18</version>
<scope>provided</scope>
</dependency>

Expand Down
5 changes: 3 additions & 2 deletions .../frontier/api/annotation/processor/annotation/service/FrontierMessageProducerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,9 @@ public FrontierMessageProducerService(
this.frontierAMQPProducer = frontierAMQPProducer;
}

public Object process(final Guarantee guarantee,
final FrontierApiIdentity identity,
public Object process(
Guarantee guarantee,
FrontierApiIdentity identity,
Set<Object> params) {
return guarantee.accept(this, identity, params);
}
Expand Down
2 changes: 1 addition & 1 deletion ...or/src/main/java/com/frontier/api/annotation/processor/api/amqp/FrontierAMQPConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ public FrontierAMQPConsumer(

@RabbitListener(queues = "${frontier-rabbitmq-queue-name}")
public void consumeMessage(String jsonPayload) throws FrontierRecoverableException {
Optional<FrontierApiRequestMessage> bodyOpt = Optional.empty();
Optional<FrontierApiRequestMessage> bodyOpt;
try {
bodyOpt = Optional
.ofNullable(new ObjectMapper().readValue(jsonPayload, FrontierApiRequestMessage.class));
Expand Down
44 changes: 11 additions & 33 deletions ...or/src/main/java/com/frontier/api/annotation/processor/api/amqp/FrontierAMQPProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,67 +1,45 @@
package com.frontier.api.annotation.processor.api.amqp;

import static com.frontier.api.annotation.processor.annotation.immutables.Guarantee.ASYNCHRONOUS;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.frontier.api.annotation.processor.api.FrontierAPIInterface;
import com.frontier.api.annotation.processor.api.immutables.FrontierApiIdentity;
import com.frontier.api.annotation.processor.api.immutables.FrontierApiRequestMessage;
import com.frontier.api.annotation.processor.exception.FrontierUnrecoverableException;
import com.frontier.api.annotation.processor.register.FrontierRegisterService;
import com.frontier.api.annotation.processor.register.immutables.FrontierDestination;
import com.frontier.api.annotation.processor.register.immutables.FrontierRegisterNode;
import java.util.Set;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

@Service
public class FrontierAMQPProducer implements
FrontierAPIInterface<Void, FrontierApiIdentity, Set<Object>> {

private final RabbitTemplate rabbitTemplate;
@Value("${frontier-queue:FRONTIER_QUEUE}")
private String FRONTIER_QUEUE;

private final FrontierRegisterService frontierRegisterService;
private final RabbitTemplate rabbitTemplate;

public FrontierAMQPProducer(RabbitTemplate rabbitTemplate,
FrontierRegisterService frontierRegisterService) {
public FrontierAMQPProducer(
RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
this.frontierRegisterService = frontierRegisterService;
}

public void produceMessage(
FrontierApiRequestMessage requestBody,
String frontierQueueName) {
try {
String json = new ObjectMapper().writeValueAsString(requestBody);
rabbitTemplate.convertAndSend(frontierQueueName, "", json);
} catch (JsonProcessingException e) {
e.printStackTrace();
}
}

@Override
public Void produceMessage(FrontierApiIdentity identity,
public Void produceMessage(
FrontierApiIdentity identity,
Set<Object> params) {

Pair<FrontierDestination, FrontierRegisterNode> discoveredDestination =
this.frontierRegisterService
.resolveFrontierDestination(identity.getBeanName(),
identity.getMethodName(),
ASYNCHRONOUS)
.orElseThrow(() -> new FrontierUnrecoverableException("Cache Miss."));

FrontierApiRequestMessage frontierApiRequestMessage = FrontierApiRequestMessage.builder()
.beanName(discoveredDestination.getRight().getBeanName())
.methodName(discoveredDestination.getRight().getMethodName())
.beanName(identity.getBeanName())
.methodName(identity.getMethodName())
.methodParams(params)
.build();

try {
String messagePayload = new ObjectMapper().writeValueAsString(frontierApiRequestMessage);
rabbitTemplate
.convertAndSend(discoveredDestination.getLeft().getName(), "", messagePayload);
.convertAndSend(FRONTIER_QUEUE, "", messagePayload);
} catch (JsonProcessingException e) {
throw new FrontierUnrecoverableException("Error publishing RabbitMQ message.");
}
Expand Down
32 changes: 6 additions & 26 deletions .../src/main/java/com/frontier/api/annotation/processor/api/rest/FrontierRestController.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,40 +1,30 @@
package com.frontier.api.annotation.processor.api.rest;

import static com.frontier.api.annotation.processor.annotation.immutables.Guarantee.SYNCHRONOUS;

import com.frontier.api.annotation.processor.annotation.service.FrontierRepositoryCacheService;
import com.frontier.api.annotation.processor.api.FrontierAPIInterface;
import com.frontier.api.annotation.processor.api.immutables.FrontierApiIdentity;
import com.frontier.api.annotation.processor.api.immutables.FrontierApiRequestMessage;
import com.frontier.api.annotation.processor.api.immutables.FrontierApiResponseMessage;
import com.frontier.api.annotation.processor.exception.FrontierRecoverableException;
import com.frontier.api.annotation.processor.exception.FrontierUnrecoverableException;
import com.frontier.api.annotation.processor.executor.FrontierExecutor;
import com.frontier.api.annotation.processor.register.FrontierRegisterService;
import com.frontier.api.annotation.processor.register.immutables.FrontierDestination;
import com.frontier.api.annotation.processor.register.immutables.FrontierRegisterNode;
import java.util.Set;
import java.util.function.Function;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.context.ApplicationContext;
import org.springframework.data.repository.Repository;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestTemplate;

@RestController
public class FrontierRestController implements
FrontierAPIInterface<FrontierApiResponseMessage, FrontierApiIdentity, Set<Object>> {

private final static String FRONTIER_ENDPOINT = "/api/frontier/";
private final static Function<Pair<FrontierDestination, FrontierRegisterNode>, String> FRONTIER_ENDPOINT_BUILDER_FUNC =
(a) -> a.getLeft().getName() + FRONTIER_ENDPOINT + a.getRight().getBeanName();
private final static String NODE_ENDPOINT = "/api/frontier/";

private final FrontierRepositoryCacheService frontierRepositoryCacheService;
private final FrontierRegisterService frontierRegisterService;
Expand All @@ -52,29 +42,19 @@ public FrontierRestController(
public FrontierApiResponseMessage produceMessage(
FrontierApiIdentity identity, Set<Object> params) {

Pair<FrontierDestination, FrontierRegisterNode> discoveredDestination =
this.frontierRegisterService
.resolveFrontierDestination(identity.getBeanName(),
identity.getMethodName(),
SYNCHRONOUS)
.orElseThrow(() -> new FrontierUnrecoverableException("Cache Miss."));

RestTemplate restTemplate = new RestTemplate();

FrontierApiRequestMessage frontierApiRequestMessage = FrontierApiRequestMessage.builder()
.beanName(discoveredDestination.getRight().getBeanName())
.methodName(discoveredDestination.getRight().getMethodName())
.beanName(identity.getBeanName())
.methodName(identity.getMethodName())
.methodParams(params)
.build();

ResponseEntity<FrontierApiResponseMessage> response = restTemplate
.postForEntity(FRONTIER_ENDPOINT_BUILDER_FUNC.apply(discoveredDestination),
frontierApiRequestMessage, FrontierApiResponseMessage.class);
ResponseEntity<FrontierApiResponseMessage> response = this.frontierRegisterService
.proxyFrontierRequest(frontierApiRequestMessage);

return response.getBody();
}

@RequestMapping(FRONTIER_ENDPOINT + "/**")
@RequestMapping(NODE_ENDPOINT + "/**")
public ResponseEntity<FrontierApiResponseMessage> frontierHandler(
HttpServletRequest request,
HttpServletResponse response,
Expand Down
2 changes: 1 addition & 1 deletion ...va/com/frontier/api/annotation/processor/api/rest/FrontierRestCustomExceptionHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,9 @@

import static org.springframework.http.HttpStatus.INTERNAL_SERVER_ERROR;

import com.frontier.api.annotation.processor.api.immutables.FrontierApiResponseMessage;
import com.frontier.api.annotation.processor.exception.FrontierRecoverableException;
import com.frontier.api.annotation.processor.exception.FrontierUnrecoverableException;
import com.frontier.api.annotation.processor.api.immutables.FrontierApiResponseMessage;
import java.util.Optional;
import org.springframework.http.HttpHeaders;
import org.springframework.http.ResponseEntity;
Expand Down
Loading