HTTP統合フローからSpring Reactor Fluxを作成するにはどうすればいいですか？

Question

私はメッセージがHTTPエンドポイントではなく、JMSキューから来る1つの違いで、この1 How to create a Spring Reactor Flux from a ActiveMQ queue?

に非常に似て質問があります。問題は、Message Channelが何らかの理由で実装されていないか、Flux.from（）によって取得されていないことです。ログエントリはGenericMessageがパス変数としてペイロードを持つHttp Integrationフローから作成されたものの、チャネルにエンキュー/パブリッシュされないことを示していますか？試しました.channel(MessageChannels.queue())と.channel(MessageChannels.publishSubscribe()) 違いはありません、イベントストリームは空です。ここでは、コードは次のとおりです。

@Bean 
public Publisher<Message<String>> httpReactiveSource() { 
     return IntegrationFlows. 
       from(Http.inboundChannelAdapter("/eventmessage/{id}") 
         .requestMapping(r -> r 
         .methods(HttpMethod.POST)                     
         ) 
         .payloadExpression("#pathVariables.id") 
         )       
         .channel(MessageChannels.queue()) 
         .log(LoggingHandler.Level.DEBUG)        
         .log() 
         .toReactivePublisher(); 
    } 


@GetMapping(value="eventmessagechannel/{id}", produces=MediaType.TEXT_EVENT_STREAM_VALUE) 
public Flux<String> eventMessages(@PathVariable String id){  
    return Flux.from(httpReactiveSource())    
      .map(Message::getPayload); 

} 

UPDATE1：

build.gradle

buildscript { 
    ext { 
     springBootVersion = '2.0.0.M2' 
    } 
    repositories { 
     mavenCentral() 
     maven { url "https://repo.spring.io/snapshot" } 
     maven { url "https://repo.spring.io/milestone" } 
    } 
    dependencies { 
     classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}") 
    } 
} 

apply plugin: 'java' 
apply plugin: 'eclipse' 
apply plugin: 'org.springframework.boot' 
apply plugin: 'io.spring.dependency-management' 

version = '0.0.1-SNAPSHOT' 
sourceCompatibility = 1.8 

repositories { 
    mavenCentral() 
    maven { url "https://repo.spring.io/snapshot" } 
    maven { url "https://repo.spring.io/milestone" } 
} 


dependencies { 
    compile('org.springframework.boot:spring-boot-starter-freemarker') 
    compile('org.springframework.boot:spring-boot-starter-integration') 
    compile('org.springframework.boot:spring-boot-starter-web') 
    compile('org.springframework.boot:spring-boot-starter-webflux') 
    compile('org.springframework.integration:spring-integration-http') 
    testCompile('org.springframework.boot:spring-boot-starter-test') 
    testCompile('io.projectreactor:reactor-test') 

} 

アップデート2

それは@SpringBootApplicationと@RestControllerは、1つのファイルで定義されているときに動作しますが、ときに動作を停止@SpringBootApplicationと@RestControllerは別々のファイルにあります。

TestApp.java

package com.example; 

import org.springframework.boot.SpringApplication; 
import org.springframework.boot.autoconfigure.SpringBootApplication; 


@SpringBootApplication 
public class TestApp { 
    public static void main(String[] args) { 
      SpringApplication.run(TestApp.class, args); 
    } 
} 

TestController.java

package com.example.controller; 


import org.springframework.context.annotation.Bean; 
import org.reactivestreams.Publisher; 
import org.springframework.http.HttpMethod; 
import org.springframework.http.MediaType; 
import org.springframework.integration.dsl.IntegrationFlows; 
import org.springframework.integration.dsl.channel.MessageChannels; 
import org.springframework.integration.handler.LoggingHandler; 
import org.springframework.integration.http.dsl.Http; 
import org.springframework.messaging.Message; 
import org.springframework.web.bind.annotation.RestController; 



import org.springframework.web.bind.annotation.GetMapping; 
import reactor.core.publisher.Flux; 



@RestController 
public class TestController { 
    @Bean 
     public Publisher<Message<String>> httpReactiveSource() { 
      return IntegrationFlows. 
        from(Http.inboundChannelAdapter("/message/{id}") 
          .requestMapping(r -> r 
            .methods(HttpMethod.POST) 
          ) 
          .payloadExpression("#pathVariables.id") 
        ) 
        .channel(MessageChannels.queue()) 
        .toReactivePublisher(); 
     } 

     @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE) 
     public Flux<String> eventMessages() { 
      return Flux.from(httpReactiveSource()) 
        .map(Message::getPayload); 
     } 

} 

Answer 1

これがうまく私の作品：

@SpringBootApplication 
@RestController 
public class SpringIntegrationSseDemoApplication { 

    public static void main(String[] args) { 
     SpringApplication.run(SpringIntegrationSseDemoApplication.class, args); 
    } 

    @Bean 
    public Publisher<Message<String>> httpReactiveSource() { 
     return IntegrationFlows. 
       from(Http.inboundChannelAdapter("/message/{id}") 
         .requestMapping(r -> r 
           .methods(HttpMethod.POST) 
         ) 
         .payloadExpression("#pathVariables.id") 
       ) 
       .channel(MessageChannels.queue()) 
       .toReactivePublisher(); 
    } 

    @GetMapping(value = "/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE) 
    public Flux<String> eventMessages() { 
     return Flux.from(httpReactiveSource()) 
       .map(Message::getPayload); 
    } 

} 

私はPOMでこの依存関係を持っている：

<parent> 
    <groupId>org.springframework.boot</groupId> 
    <artifactId>spring-boot-starter-parent</artifactId> 
    <version>2.0.0.BUILD-SNAPSHOT</version> 
    <relativePath/> <!-- lookup parent from repository --> 
</parent> 

<dependencies> 
    <dependency> 
     <groupId>org.springframework.boot</groupId> 
     <artifactId>spring-boot-starter-integration</artifactId> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework.boot</groupId> 
     <artifactId>spring-boot-starter-web</artifactId> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework.boot</groupId> 
     <artifactId>spring-boot-starter-webflux</artifactId> 
    </dependency> 
    <dependency> 
     <groupId>org.springframework.integration</groupId> 
     <artifactId>spring-integration-http</artifactId> 
    </dependency> 

    <dependency> 
     <groupId>org.springframework.boot</groupId> 
     <artifactId>spring-boot-starter-test</artifactId> 
     <scope>test</scope> 
    </dependency> 
</dependencies> 

<build> 
    <plugins> 
     <plugin> 
      <groupId>org.springframework.boot</groupId> 
      <artifactId>spring-boot-maven-plugin</artifactId> 
     </plugin> 
    </plugins> 
</build> 

私は、アプリケーションを実行する2つの端子があります。SSESを聞くこと

curl http://localhost:8080/events 

を。

そして、私はこれを実行秒1で：だから

curl -X POST http://localhost:8080/message/foo 

curl -X POST http://localhost:8080/message/bar 

curl -X POST http://localhost:8080/message/666 

、最初の端末は次のように応答します。

​​

注意、我々はspring-boot-starter-webflux依存関係を必要としません。 FluxからSSEまでは、サーブレットコンテナ上の通常のMVCでうまく動作します。

Spring Integrationは、すぐにWebFluxもサポートします：https://jira.spring.io/browse/INT-4300。

IntegrationFlows 
    .from(Http.inboundReactiveGateway("/sse") 
         .requestMapping(m -> m.produces(MediaType.TEXT_EVENT_STREAM_VALUE))) 

、完全には、任意のサーブレットコンテナ依存せずに、単にだけWebFluxを頼る：だから、あなたはそこのようなものを設定することができるようになります。