标签: reactive

我已经重构了我的代码以使用 spring webflux 但现在@Valid停止工作。它不验证请求正文。

@PostMapping(value = "/getContactInfo",produces = "application/json",consumes = "application/json")
public Flux<UserContactsModel> getUserContacts(@Valid @RequestBody Mono<LoginModel> loginDetail) {

    loginDetail.log();
    return contactInfoService
        .getUserContacts(loginDetailApiMapper.loginModelMonoToLoginBoMono(loginDetail))
        .flatMapIterable(
            userContactsBO -> contactInfoMapper.userContactBoToModelList(userContactsBO));
}

我收到 200 OK 代替了我从控制器建议返回的 Bad request。

编辑1：

   import javax.validation.constraints.NotNull;
   import javax.validation.constraints.Pattern;

    public class LoginModel implements Serializable {

      private String clientId;

      @Pattern(regexp = "^[a-zA-Z0-9]*$", message = "Login ID is invalid")
      @NotNull
      private String loginId;
    }

更新1： 像这样更改代码并在类级别添加@Validated之后

@RestController
@Validated
public class ContactInfoController implements ContactInfoApi {
public Flux<UserContactsModel> getUserContacts(@RequestBody  Mono<@Valid  LoginModel> loginDetail) { …

我有两种方法.
主要方法:

@PostMapping("/login")
public Mono<ResponseEntity<ApiResponseLogin>> loginUser(@RequestBody final LoginUser loginUser) {
    return socialService.verifyAccount(loginUser)
            .flatMap(socialAccountIsValid -> {
                if (socialAccountIsValid) {
                    return this.userService.getUserByEmail(loginUser.getEmail())
                            .switchIfEmpty(insertUser(loginUser))
                            .flatMap(foundUser -> updateUser(loginUser, foundUser))
                            .map(savedUser -> {
                                String jwts = jwt.createJwts(savedUser.get_id(), savedUser.getFirstName(), "user");
                                return new ResponseEntity<>(HttpStatus.OK);
                            });
                } else {
                    return Mono.just(new ResponseEntity<>(HttpStatus.UNAUTHORIZED));
                }
            });

}

这个被调用的方法(该服务调用外部api):

public Mono<User> getUserByEmail(String email) {
    UriComponentsBuilder builder = UriComponentsBuilder
            .fromHttpUrl(USER_API_BASE_URI)
            .queryParam("email", email);
    return this.webClient.get()
            .uri(builder.toUriString())
            .exchange()
            .flatMap(resp -> {
                if (Integer.valueOf(404).equals(resp.statusCode().value())) {
                    return Mono.empty();
                } else {
                    return resp.bodyToMono(User.class);
                }
            });
} …

我期待从Spring反应webclient进行SOAP调用.我找不到任何文件.想知道这个方法会是什么.现在我在想

1. 在单独的线程池上使用JAXB构造SOAP消息
2. 通过webclient将其转换为字符串来进行调用
3. 在返回单独的tp的路上,使用jaxb转换回java.

有什么缺点和其他方法？

我使用Vert.x工具包来创建反应式应用程序,支持MySQL和Postgres等关系数据库.我知道Spring为像Cassandra和Mongo这样的NoSQL DB提供了反应支持,但是他们是否愿意为关系数据库提供相同的支持？

我使用 Web-flux/Reactive 和 Webclient，在 tomcat 和 spring-boot 上运行它。

一切正常。我阅读了很多关于它的信息。问题似乎是每当你使用 webclient 时，你必须返回或使用响应，否则它会关闭连接而你还没有使用它，你会看到很多日志消息说the connection close prematurely，如果我有404 状态代码是一个错误的场景，我可以使用OnStatus并抛出异常，但我的情况是：当上游服务返回 404 时，我必须手动返回单声道空。所以我不使用来自 Web 客户端请求的响应，我只是使用来自 .exchange() 的 ClientResponse 来检查状态并处理它。我最初的问题是日志消息，因为这只是“垃圾”，您不希望在日志消息中看到很多。我在某处读到，如果发生这种情况，连接也不能重复使用，所以这听起来很糟糕，但我不知道......如果没有找到，我只会收到这条消息，如果响应是 200它返回对象并且不打印日志消息。

我尝试使用 clientResponse.BodyToMono(Void.Class) 但它也不起作用。保持出现的日志消息

 @Bean
  public WebClient webClient(
      @Value("${http.client.connection-timeout-millis}") final int connectionTimeoutMillis,
      @Value("${http.client.socket-timeout-millis}") final int socketTimeoutMillis,
      @Value("${http.client.wire-tap-enabled}") final boolean wireTapEnabled,
      final ObjectMapper objectMapper) {

    Consumer<Connection> doOnConnectedConsumer = connection ->
        connection
            .addHandler(new ReadTimeoutHandler(socketTimeoutMillis, MILLISECONDS))
            .addHandler(new WriteTimeoutHandler(connectionTimeoutMillis, MILLISECONDS));

    TcpClient tcpClient = TcpClient.newConnection()
        .wiretap(wireTapEnabled)
        .option(CONNECT_TIMEOUT_MILLIS, connectionTimeoutMillis)
        .doOnConnected(doOnConnectedConsumer);

    return WebClient.builder()
        .clientConnector(new ReactorClientHttpConnector(HttpClient.from(tcpClient).compress(true)))
        .exchangeStrategies(customExchangeStrategies(objectMapper))
        .build();
  } 

// .......... …

我想从一个可观察者中获取最后三个元素.假设我的时间表如下所示:

--a---b-c---d---e---f-g-h-i------j->

哪里: a, b, c, d, e, f, g, h, i, j are emitted values

每当发出一个新值时我想立即得到它,所以它看起来像这样:

[a]
[a, b]
[a, b, c]
[b, c, d]
[c, d, e]
[d, e, f]
[e, f, g]
[f, g, h]
... and so on

我认为这非常有用.想象一下,建立一个你想要显示10条最后消息的聊天.每当有新消息出现时,您都希望更新视图.

我的尝试:演示

我对 Reactive Stream 有点陌生，所以在使用Spring Webflux和Reactor时遇到了一个问题。

我制作了一个如下所示的片段：

@RestController
public class TestController {
    @GetMapping("responsebody/flux")
    public Flux<String> tt2() {
        return Flux.range(1, 5)
            .delayElements(Duration.ofMillis(1000))
            .map(l -> "hi");
    }
}

而且，有趣的是，镶边分别显示序列中的每个元素，而不是当我仅使用浏览器请求时一次公开所有元素。（但开发工具立即显示全身）

但我想知道，即使 HTTP 1 只使用一个连接，并且服务器发送的数据放在 HTTP 协议的正文中，它是如何工作的。客户端如何知道哪个元素分隔每个元素以及序列何时完成？如果客户端还没有准备好使用反应流怎么办？

我不需要任何使用反应式库的代码，但想知道协议是如何工作的。

我有一个针对.Net 4.5.2的多项目Visual Studio解决方案.在其中一个项目(WPF应用程序)中,我使用nuget添加System.Reactive版本3.0.1000.0包,然后添加ReactiveUI 7.0.0.0包.

在WPF应用程序使用的另一个项目库中,我只添加了System.Reactive版本3.0.1000.0包.

ReactiveUI包似乎依赖于一组旧的反应包(RX-Core2.2.5等).我可以告诉你,因为WPF应用程序项目文件中的HintPaths指向诸如packages\Rx-Core.2.2.5\lib \net45\System.Reactive.Core.dll之类的位置

当我构建并运行应用程序时,我得到一个FileLoadException,因为至少有一个项目试图使用错误的dll版本.以下是典型的....

System.IO.FileLoadException occurred
  HResult=0x80131040
  Message=Could not load file or assembly 'System.Reactive.Linq, Version=3.0.1000.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35' or one of its dependencies. The located assembly's manifest definition does not match the assembly reference. (Exception from HRESULT: 0x80131040)

我可以通过将解决方案中的所有System.Reactive软件包降级到2.2.5来解决这个问题,但这似乎是一个非常旧的版本(2014).

为什么ReactiveUI会依赖System.Reactive的v2.2.5？有没有办法改变这种行为,以便我可以在整个解决方案中使用最新版本的System.Reactive？

现在有两种方法可以在Spring 5中公开http端点.

1. 控制器:通过制作休息控​​制器.

   @RestController
   @RequestMapping("persons")
   public class PersonController { 
   
       @Autowired
       private PersonRepo repo;
   
       @GetMapping("/{id}")
       public Mono<Person> personById(@PathVariable String id){
           retrun repo.findById(id);
       }
   }
   

   Run Code Online (Sandbox Code Playgroud)

2. 路由器:通过路由器.例如:

   @Bean
   public RouterFunction<ServerResponse> personRoute(PersonRepo repo) {
       return route(GET("/persons/{id}"), req -> Mono.justOrEmpty(req.pathVariable("id"))                                             
                                                    .flatMap(repo::getById)
                                                    .flatMap(p -> ok().syncBody(p))
                                                    .switchIfEmpty(notFound().build()));
   }
   

   Run Code Online (Sandbox Code Playgroud)

使用任何一种方法有任何性能差异吗？从头开始编写应用程序时,我应该使用哪一个.

我最近遇到过"反应套接字"一词.到目前为止,我曾经认为websockets是完全成熟的异步风格的方法.

什么是无功插座.

这个链接(http://reactivesocket.io/)甚至讨论了websockets的比较.