-
Notifications
You must be signed in to change notification settings - Fork 79
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Refactor: Remove publishing code duplication (#425)
* refactor: remove publishing code duplication * refactor: remove publishing code duplication * test: fix test and dependencies after refactor * test: fix test and dependencies after refactor
- Loading branch information
Showing
23 changed files
with
245 additions
and
285 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
50 changes: 50 additions & 0 deletions
50
...in/java/io/github/stavshamir/springwolf/asyncapi/controller/PublishingBaseController.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package io.github.stavshamir.springwolf.asyncapi.controller; | ||
|
|
||
| import io.github.stavshamir.springwolf.asyncapi.controller.dtos.MessageDto; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.beans.factory.InitializingBean; | ||
| import org.springframework.http.HttpStatus; | ||
| import org.springframework.http.ResponseEntity; | ||
| import org.springframework.web.bind.annotation.PostMapping; | ||
| import org.springframework.web.bind.annotation.RequestBody; | ||
| import org.springframework.web.bind.annotation.RequestParam; | ||
|
|
||
| /** | ||
| * Used in plugins with publishing enabled. | ||
| * Located in springwolf-core to allow sharing of code | ||
| */ | ||
| @RequiredArgsConstructor | ||
| @Slf4j | ||
| public abstract class PublishingBaseController implements InitializingBean { | ||
|
|
||
| private final PublishingPayloadCreator publishingPayloadCreator; | ||
|
|
||
| protected abstract boolean isEnabled(); | ||
|
|
||
| protected abstract void publishMessage(String topic, MessageDto message, Object payload); | ||
|
|
||
| @PostMapping("/publish") | ||
| public ResponseEntity<String> publish(@RequestParam String topic, @RequestBody MessageDto message) { | ||
| if (!isEnabled()) { | ||
| String errorMessage = "Publishing using %s is not enabled - message will not be published" | ||
| .formatted(this.getClass().getSimpleName()); | ||
| log.warn(errorMessage); | ||
| return ResponseEntity.status(HttpStatus.NOT_FOUND).body(errorMessage); | ||
| } | ||
|
|
||
| PublishingPayloadCreator.Result result = publishingPayloadCreator.createPayloadObject(message); | ||
| if (result.payload() != null) { | ||
| publishMessage(topic, message, result.payload()); | ||
| return ResponseEntity.ok().build(); | ||
| } | ||
| return ResponseEntity.badRequest().body(result.errorMessage()); | ||
| } | ||
|
|
||
| @Override | ||
| public void afterPropertiesSet() { | ||
| log.debug( | ||
| "Message publishing via %s is active.".formatted(this.getClass().getSimpleName())); | ||
| } | ||
| } |
52 changes: 52 additions & 0 deletions
52
...in/java/io/github/stavshamir/springwolf/asyncapi/controller/PublishingPayloadCreator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,52 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package io.github.stavshamir.springwolf.asyncapi.controller; | ||
|
|
||
| import com.fasterxml.jackson.core.JsonProcessingException; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import io.github.stavshamir.springwolf.asyncapi.controller.dtos.MessageDto; | ||
| import io.github.stavshamir.springwolf.schemas.SchemasService; | ||
| import io.swagger.v3.oas.models.media.Schema; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
|
|
||
| import java.text.MessageFormat; | ||
|
|
||
| /** | ||
| * Used in plugins with publishing enabled. | ||
| * Located in springwolf-core to allow sharing of code | ||
| */ | ||
| @RequiredArgsConstructor | ||
| @Slf4j | ||
| public class PublishingPayloadCreator { | ||
|
|
||
| private final SchemasService schemasService; | ||
| private final ObjectMapper objectMapper; | ||
|
|
||
| public Result createPayloadObject(MessageDto message) { | ||
|
|
||
| String messagePayloadType = message.getPayloadType(); | ||
| for (Schema<?> value : schemasService.getDefinitions().values()) { | ||
| String schemaPayloadType = value.getName(); | ||
| // security: match against user input, but always use our controlled data from the DefaultSchemaService | ||
| if (schemaPayloadType.equals(messagePayloadType)) { | ||
| try { | ||
| Class<?> payloadClass = Class.forName(schemaPayloadType); | ||
| Object payload = objectMapper.readValue(message.getPayload(), payloadClass); | ||
| return new Result(payload, null); | ||
| } catch (ClassNotFoundException | JsonProcessingException ex) { | ||
| String errorMessage = MessageFormat.format( | ||
| "Unable to create payload {0} from data: {1}", schemaPayloadType, message.getPayload()); | ||
| log.info(errorMessage, ex); | ||
| return new Result(null, errorMessage); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| String errorMessage = MessageFormat.format( | ||
| "Specified payloadType {0} is not a registered springwolf schema.", messagePayloadType); | ||
| log.info(errorMessage); | ||
| return new Result(null, errorMessage); | ||
| } | ||
|
|
||
| public record Result(Object payload, String errorMessage) {} | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
17 changes: 17 additions & 0 deletions
17
...src/test/java/io/github/stavshamir/springwolf/fixtures/ObjectMapperTestConfiguration.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package io.github.stavshamir.springwolf.fixtures; | ||
|
|
||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; | ||
| import org.springframework.boot.test.context.TestConfiguration; | ||
| import org.springframework.context.annotation.Bean; | ||
|
|
||
| @TestConfiguration | ||
| public class ObjectMapperTestConfiguration { | ||
|
|
||
| @ConditionalOnMissingBean | ||
| @Bean | ||
| public ObjectMapper objectMapper() { | ||
| return new ObjectMapper(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
71 changes: 12 additions & 59 deletions
71
...in/java/io/github/stavshamir/springwolf/asyncapi/controller/SpringwolfAmqpController.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,80 +1,33 @@ | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
| package io.github.stavshamir.springwolf.asyncapi.controller; | ||
|
|
||
| import com.fasterxml.jackson.core.JsonProcessingException; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import io.github.stavshamir.springwolf.asyncapi.controller.dtos.MessageDto; | ||
| import io.github.stavshamir.springwolf.producer.SpringwolfAmqpProducer; | ||
| import io.github.stavshamir.springwolf.schemas.SchemasService; | ||
| import io.swagger.v3.oas.models.media.Schema; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.springframework.beans.factory.InitializingBean; | ||
| import org.springframework.http.HttpStatus; | ||
| import org.springframework.web.bind.annotation.PostMapping; | ||
| import org.springframework.web.bind.annotation.RequestBody; | ||
| import org.springframework.web.bind.annotation.RequestMapping; | ||
| import org.springframework.web.bind.annotation.RequestParam; | ||
| import org.springframework.web.bind.annotation.RestController; | ||
| import org.springframework.web.server.ResponseStatusException; | ||
|
|
||
| import java.text.MessageFormat; | ||
|
|
||
| @Slf4j | ||
| @RestController | ||
| @RequestMapping("/springwolf/amqp") | ||
| @RequiredArgsConstructor | ||
| public class SpringwolfAmqpController implements InitializingBean { | ||
|
|
||
| private final SchemasService schemasService; | ||
| @Slf4j | ||
| public class SpringwolfAmqpController extends PublishingBaseController { | ||
|
|
||
| private final SpringwolfAmqpProducer producer; | ||
|
|
||
| private final ObjectMapper objectMapper; | ||
|
|
||
| @PostMapping("/publish") | ||
| public void publish(@RequestParam String topic, @RequestBody MessageDto message) { | ||
| if (!producer.isEnabled()) { | ||
| log.warn("AMQP producer is not enabled - message will not be published"); | ||
| throw new ResponseStatusException(HttpStatus.NOT_FOUND, "AMQP producer is not enabled"); | ||
| } | ||
|
|
||
| boolean foundDefinition = false; | ||
| String messagePayloadType = message.getPayloadType(); | ||
| for (Schema<?> value : schemasService.getDefinitions().values()) { | ||
| String schemaPayloadType = value.getName(); | ||
| // security: match against user input, but always use our controlled data from the DefaultSchemaService | ||
| if (schemaPayloadType.equals(messagePayloadType)) { | ||
| publishMessage(topic, message, schemaPayloadType); | ||
|
|
||
| foundDefinition = true; | ||
| break; | ||
| } | ||
| } | ||
|
|
||
| if (!foundDefinition) { | ||
| throw new ResponseStatusException( | ||
| HttpStatus.BAD_REQUEST, "Specified payloadType is not a registered springwolf schema."); | ||
| } | ||
| public SpringwolfAmqpController( | ||
| PublishingPayloadCreator publishingPayloadCreator, SpringwolfAmqpProducer producer) { | ||
| super(publishingPayloadCreator); | ||
| this.producer = producer; | ||
| } | ||
|
|
||
| private void publishMessage(String topic, MessageDto message, String schemaPayloadType) { | ||
| try { | ||
| Class<?> payloadClass = Class.forName(schemaPayloadType); | ||
| Object payload = objectMapper.readValue(message.getPayload(), payloadClass); | ||
|
|
||
| log.debug("Publishing to amqp queue {}: {}", topic, message.getPayload()); | ||
| producer.send(topic, payload); | ||
| } catch (ClassNotFoundException | JsonProcessingException ex) { | ||
| throw new ResponseStatusException( | ||
| HttpStatus.BAD_REQUEST, | ||
| MessageFormat.format( | ||
| "Unable to create payload {0} from data: {1}", schemaPayloadType, message.getPayload())); | ||
| } | ||
| @Override | ||
| protected boolean isEnabled() { | ||
| return producer.isEnabled(); | ||
| } | ||
|
|
||
| @Override | ||
| public void afterPropertiesSet() { | ||
| log.debug("Message publishing via " + this.getClass().getSimpleName() + " is active."); | ||
| protected void publishMessage(String topic, MessageDto message, Object payload) { | ||
| log.debug("Publishing to amqp queue {}: {}", topic, message.getPayload()); | ||
| producer.send(topic, payload); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.