Alpakka JMS 거래

Question

나는 Oracle AQ에서 데이터를 대기열에서 제외하기 위해 Alpakka 및 그 JMS 커넥터로 게임을하고 있습니다. 아래의 매우 기본적인 구현 방법을 따라 가면 this 가이드가 될 수 있습니다.

제 질문은 어떻게 처리 할 수 ​​있습니까? 예외가 발생하면 내 메시지가 손실되지 않도록 보장 할 수 있습니다. 이 PL/SQL 인 경우

object ConsumerApp extends App { 
    implicit val system: ActorSystem = ActorSystem("actor-system") 
    implicit val materializer: ActorMaterializer = ActorMaterializer() 

    val connectionFactory = AQjmsFactory.getConnectionFactory(getOracleDataSource()) 

    val out = JmsSource.textSource(
     JmsSourceSettings(connectionFactory).withQueue("My_Queue") 
    ) 

    val sink = Sink.foreach { message: String => 
     println("in sink: " + message) 
     throw new Exception("") // !!! MESSAGE IS LOST !!! 
    } 

    out.runWith(sink, materializer) 
} 

,이 솔루션은 다음과 같이 될 것이다 :

DECLARE 
    dequeue_options   DBMS_AQ.DEQUEUE_OPTIONS_T; 
    message_properties   DBMS_AQ.MESSAGE_PROPERTIES_T; 
    message_handle    RAW (44); 
    msg      SYS.AQ$_JMS_TEXT_MESSAGE; 
BEGIN 
    DBMS_AQ.dequeue (
     queue_name   => 'My_Queue', 
     dequeue_options  => dequeue_options, 
     message_properties => message_properties, 
     payload    => msg, 
     msgid    => message_handle 
); 

    -- do something with the message 

    COMMIT; 
END; 

Answer 1

기본 동작 스트림 단계는 전체 스트림을 종료하는 것입니다 실패 할 때. 스트림에서 원하는 방식을 결정해야합니다. handle errors 예를 들어, 한 가지 접근법은 백 오프 전략을 사용하여 스트림을 restart으로 만드는 것입니다.

또한 Alpakka JMS 커넥터를 사용하고 있으므로 acknowledgement mode을 ClientAcknowledge (Alpakka 0.15부터 사용 가능)으로 설정하십시오. 이 구성을 사용하면 승인되지 않은 메시지를 JMS 소스를 통해 다시 전달할 수 있습니다. 예 :

val jmsSource: Source[Message, NotUsed] = JmsSource(
    JmsSourceSettings(connectionFactory) 
    .withQueue("My_Queue") 
    .withAcknowledgeMode(AcknowledgeMode.ClientAcknowledge) 
) 

val result = jmsSource 
    .map { 
    case textMessage: TextMessage => 
     val text = textMessage.getText 
     textMessage.acknowledge() 
     text 
    } 
    .runForeach(println)