HTTP Inicio Rápido del Flujo de Recepción

Al recibir mensajes de sistemas externos a través de HTTP es posible aprovechar los beneficios del marco de conectores sin necesariamente utilizar un receive connector. Esta página demuestra esto utilizando Spring REST Controladores y es posible mediante el HttpReceiveFlow y HttpReceiveFlowService componentes.

Dependencias

Antes de construir un HTTP Recibir Flujo/HTTP Reciba el Servicio, el connector-http la biblioteca debe ser incluida como una dependencia.

 <dependency>
    <groupId>com.iconsolutions.ipf.core.connector</groupId>
    <artifactId>connector-http</artifactId>
    <version>${connector.version}</version>
</dependency>

La última versión de la biblioteca de conectores se puede encontrar utilizando este Búsqueda de Nexus.

Introducción:HTTP Recibir Flujo

HTTP Los Flujos de Recepción pueden definirse para proporcionar una forma estandarizada de implementar funcionalidades como el registro, la correlación y otras funcionalidades que tradicionalmente están disponibles al utilizar un receive connector, al utilizar otro mecanismo que el receive connector para exponer un endpoint (por ejemplo, un controlador de Spring)

Patrón Constructor

 HTTP Los Flujos de Recepción se instancian utilizando el patrón de constructor.
Esto es porque HTTP Los flujos de recepción tienen muchos parámetros que configurar y la mayoría son opcionales o tienen valores predeterminados.

Veamos cómo utilizamos el patrón de constructor para instanciar un HTTP recibir flujo.

Al construir un HTTP recibir flujo lo configuramos para que sea un receptor iniciador o un receptor de respuesta. Un receptor iniciador recibe solicitudes de un sistema externo, mientras que un receptor de respuestas espera que los mensajes sean respuestas a solicitudes realizadas previamente a través de un sending connector.

Iniciando Receptor

El siguiente ejemplo demuestra las propiedades mínimas que deben ser proporcionadas al construir un flujo de recepción inicial.

HttpReceiveFlow<ExampleType> connector = HttpReceiveFlow
        .<ExampleType>builder("ExampleSystem") (1)
        .withReceiveTransportMessageConverter(converter) (3)
        .withProcessingContextExtractor(processingContextExtractor) (4)
        .withReceiveHandler(receiver) (5)
        .withActorSystem(actorSystem) (6)
        .build();
1 Establece el nombre del HTTP recibir flujo. El nombre debe representar los mensajes del sistema externo de los cuales el flujo está procesando mensajes.
2 Proporciona una implementación del ReceiveTransportMessageConverter interfaz. Toma el recibido TransportMessage y lo convierte al tipo objetivo T(ExampleType en este caso).
3 Proporciona una implementación del ProcessingContextExtractor interfaz. Este campo es lo que hace de este un conector de recepción inicial, ya que extrae (o genera) un ProcessingContext desde el mensaje en lugar de obtener uno del servicio de correlación como sería el caso en un conector que recibe respuestas.
4 Una implementación de ReceiveHandler. Aquí es donde iría la lógica de la aplicación para decidir cómo manejar las solicitudes.
5 Establece el sistema de actores utilizado en toda la aplicación.

Receptor de Respuesta

El siguiente ejemplo demuestra cómo construir un conector de recepción de respuesta mínima.

HttpReceiveFlow<ExampleType> connector = HttpReceiveFlow
        .<ExampleType>builder("connector-name") (1)
        .withReceiveTransportMessageConverter(converter) (3)
        .withCorrelationIdExtractor(correlationIdExtractor) (4)
        .withCorrelationService(correlationService) (5)
        .withReceiveHandler(receiver) (6)
        .withActorSystem(actorSystem) (7)
        .build();
1 Establezca el nombre del HTTP recibir flujo. El nombre debe representar los mensajes del sistema externo de los cuales el flujo está procesando mensajes.
2 Proporciona una implementación del ReceiveTransportMessageConverter interfaz. Toma el recibido TransportMessage y lo convierte al tipo objetivo,ExampleType en este caso.
3 Proporciona una implementación de la CorrelationIdExtractor interfaz. Toma el mensaje recibido y extrae el identificador de correlación para que podamos correlacionarlo con la solicitud original realizada a través de un sending connector.
4 Proporciona una implementación del CorrelationService interfaz. El servicio de correlación toma el identificador de correlación extraído y devuelve el ProcessingContext asociado que se utilizó cuando se envió la solicitud original a través de un sending connector.
5 Una implementación de ReceiveHandler. Aquí es donde iría la lógica de la aplicación para decidir cómo manejar las respuestas.
6 Establece el sistema de actores utilizado en toda la aplicación.

HTTP Servicio de Flujo de Recepción

Una vez que haya configurado un HTTP Recibir flujo, usted debe definir un HTTP Servicio de Flujo de Recepción bean. Este servicio es responsable de enviar las solicitudes recibidas del controlador de Spring a la Akka Streams flujo para procesar y devolver la respuesta.

A continuación se muestra un ejemplo de cómo configurar el bean

@Bean
HttpReceiveFlowService<RequestMessage> httpReceiveFlowService(ActorSystem actorSystem, HttpReceiveFlow<RequestMessage> httpReceiveFlow) {
    return new HttpReceiveFlowService<>(actorSystem, httpReceiveFlow);
}

Procesando Mensajes

Una vez que el HttpReceiveFlow y HttpReceiveFlowService beans han sido configurados, simplemente conecte el HttpReceiveFlowService en su controlador de Spring e invoque el process método, pasando la solicitud a ser procesada. Invocar este método significa que la solicitud será procesada a través de toda la funcionalidad configurada en el HttpReceiveFlow como message log ging y correlación. Opcionalmente, también puede pasar MessageHeaders como el segundo parámetro para el process método.

package com.iconsolutions.ipf.core.connector.example.app;

import com.iconsolutions.ipf.core.connector.HttpReceiveFlowService;
import com.iconsolutions.ipf.core.connector.example.model.RequestMessage;
import lombok.RequiredArgsConstructor;
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.RequestMethod;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

import java.util.Optional;

import static com.iconsolutions.ipf.core.connector.HttpReceiveFlowService.ACK;
import static com.iconsolutions.ipf.core.connector.HttpReceiveFlowService.HTTP_RECEIVE_FLOW_SERVICE_RESPONSE;
import static com.iconsolutions.ipf.core.connector.HttpReceiveFlowService.NACK;

@RestController
@RequiredArgsConstructor
public class ExampleController {

    private final HttpReceiveFlowService<RequestMessage> httpReceiveFlowService;
    @RequestMapping(value = "/submit", method = RequestMethod.POST)
    public Mono<ResponseEntity<Object>> submit(@RequestBody RequestMessage requestMessage) {
        return httpReceiveFlowService.process(requestMessage) (1)
                .map(response -> {
                    if(ACK.equals(Optional.ofNullable(response.getMessageHeaders().getHeaderMap().get(HTTP_RECEIVE_FLOW_SERVICE_RESPONSE)).map(Object::toString).orElse(NACK))) {
                        return ResponseEntity.accepted().build();
                    }
                    return ResponseEntity.internalServerError().build();
                });
    }

}
1 El HttpReceiveFlowService devuelve un Mono<TransportMessage> haciéndolo independiente de la tecnología. El mensaje contendrá un MessageHeader para indicar el éxito o el fracaso de la solicitud (ack/nack) así como los detalles de las excepciones que pueden haber ocurrido. Si el mensaje es un acuse de recibo, la carga útil contendrá el receiveContext de la solicitud original, o en el caso de un reconocimiento negativo, contendrá la carga útil de la solicitud original. En el ejemplo anterior, traducimos esta respuesta a un formato específico de Spring.ResponseEntity.

Configuración

Los valores que pueden ser configurados a través de las propiedades de configuración se muestran en la siguiente tabla.

Propiedad Descripción Ejemplo

throttle-count

Si se establece el valor, limita el rendimiento a un número específico de mensajes consumidos por unidad de tiempo. Si está configurado. Si este valor está establecido, también debe establecerse la duración del estrangulador.

10

throttle-duration

Si se establece, se utiliza junto con throttle-count para establecer la tasa máxima para consumir mensajes. Para más detalles, consulte doc.akka.io/japi/akka/2. 6/akka/stream/javadsl/Flow.html#throttle(int,java.time. Duration)

1s

mapping-parallelism

Si se establece, limita el número de concurrentes.mapping operaciones ejecutadas sobre los mensajes consumidos.

número de procesadores disponibles

receiver-parallelism-type

Define la manera en que se manejan los mensajes en paralelo.

  • ORDERED - Los mensajes se consumen en paralelo en el orden en que son recibidos, y se reconocen en el mismo orden.

  • ORDERED_PARTITIONED - los mensajes se consumen en paralelo en el orden en que son recibidos, y se reconocen en el mismo orden, pero el paralelismo para los mensajes que comparten el UnitOfWorkId está limitado a un grado configurable

  • UNORDERED - los mensajes se consumen en paralelo en el orden en que son recibidos, y se reconocen en su orden de finalización, lo que puede afectar a algunos transportes (por ejemplo,Kafka)

ORDERED_PARTITIONED

receiver-parallelism

Si se establece, limita el número de mensajes mapeados que se permiten procesar de manera concurrente.

número de procesadores disponibles

receiver-parallelism-per-partition

Solo se aplica si receiver-parallelism-type está configurado para ORDERED_PARTITIONED. Si se establece, limita el número de mensajes mapeados por UnitOfWorkId que se permiten procesar de manera concurrente. Debe ser menos que receiver-parallelism.

1

resiliency-settings

Los ajustes de resiliencia que se utilizarán al recibir. Para más detalles, consulte el Resiliencia documentación.