如何在 Spring 中使用 R2DBC 为项目实现 rsockets？

如何解决如何在 Spring 中使用 R2DBC 为项目实现 rsockets？

我正在 Spring 和 Vue 中使用响应式 websockets/rsockets 做一个通知项目。但是我一直无法将我的 ws 绑定到数据库。我被卡住了，我需要帮助。

这是我的实体（ProductDto 和 ProductRequestDto 是相同的语法）

package com.example.productservice.entity;

import org.springframework.data.annotation.Id;

import lombok.Data;
import lombok.ToString;

@Data
@ToString
public class Products {

    @Id
    private Integer id;
    private String description;
    private Integer price;
    private String subscriber;

    public Products() {

    }

    //Getters And Setters
}

这是我的仓库

package com.example.productservice.repository;

import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;

import com.example.productservice.entity.Products;

import reactor.core.publisher.Mono;

@Repository
public interface ProductRepository extends ReactiveCrudRepository<Products,Integer> {
    
    @Query  //Example of Query (Ignore this)
    ("insert into products values (null,:product,:price,:subscriber)")
    Mono<Boolean> insert (String product,int price,String subscriber);

}

这是我的服务（我留下完整的代码，因为我认为它可能对某人有用）

package com.example.productservice.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.example.productservice.dto.ProductDto;
import com.example.productservice.repository.ProductRepository;
import com.example.productservice.util.EntityDtoUtil;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

@Service
public class ProductService {
    
    @Autowired
    private ProductRepository repository;
    
    public Flux<ProductDto> getAll(){
        return this.repository.findAll()
        .map(EntityDtoUtil::toDto);
    }
    
    public Mono<ProductDto> getProductById(final int id){
            return this.repository.findById(id)
                    .map(EntityDtoUtil::toDto);
    }
    
    public Mono<ProductDto> insertProduct(Mono<ProductDto> productDtoMono){
        return productDtoMono
                .map(EntityDtoUtil::toEntity)
                .flatMap(this.repository::save)
                .map(EntityDtoUtil::toDto);
    }

    public Mono<ProductDto> updateProduct(int id,Mono<ProductDto> productDtoMono){
        return this.repository.findById(id)
                .flatMap(p -> productDtoMono
                        .map(EntityDtoUtil::toEntity)
                        .doOnNext(e -> e.setId(id)))
                .flatMap(this.repository::save)
                .map(EntityDtoUtil::toDto);
    }
    
    public Mono<Void> deleteProduct (int id) {
        return this.repository.deleteById(id);
    }

}

这是我的控制器

package com.example.productservice.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.PutMapping;
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 com.example.productservice.dto.ProductDto;
import com.example.productservice.service.ProductService;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
@RequestMapping("product")
@CrossOrigin(origins= "*",methods = {RequestMethod.GET,RequestMethod.POST})
public class ProductController {
    
    @Autowired
    private ProductService service;
    
    @GetMapping("all")
    public Flux<ProductDto> all(){
        return this.service.getAll();
    }
    
    @GetMapping("{id}")
    public Mono<ResponseEntity<ProductDto>> getProductById(int id){
        return this.service.getProductById(id)
                .map(ResponseEntity::ok)
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    
    @PostMapping
    public Mono<ProductDto> insertProductTable(@RequestBody Mono<ProductDto> productDtoMono){
        return this.service.insertProduct(productDtoMono);
    }
    
    @PutMapping("{id}")
    public Mono<ResponseEntity<ProductDto>> updateProduct(@PathVariable int id,@RequestBody Mono<ProductDto> productMonoDto) {
        return this.service.updateProduct(id,productMonoDto)
                .map(ResponseEntity::ok)
                .defaultIfEmpty(ResponseEntity.notFound().build());
    }
    
    @DeleteMapping
    public Mono<Void> deleteProduct(@PathVariable int id){
        return this.service.deleteProduct(id);
    }
    
}

到目前为止，我的原始代码以传统方式运行。我已经尝试过实现 rsockets 的方法，但我离开这里：

我在服务器端的 applicattion.properties

spring.rsocket.server.port=7000
spring.rsocket.server.transport=tcp
spring.r2dbc.url=r2dbc:mysql://localhost:3306/people2?options=DB_CLOSE_DELAY=-1;DB_CLOSE_ON_EXIT=FALSE
spring.r2dbc.username=root
spring.r2dbc.password=

ProductService 中的修改

@Service
public class ProductService {
    
    @Autowired
    private ProductRepository repository;
    
    @Autowired
    //private Sinks.Many<ProductDto> sink;
    
    @MessageMapping("repository.findAll")
    public Flux<ProductDto> getAll(){
        return this.repository.findAll()
        .map(EntityDtoUtil::toDto);
    }

ProductController 客户端的修改

@RestController
@RequestMapping("product")
@CrossOrigin(origins= "*",RequestMethod.POST})
public class ProductController {
    
    @Autowired
    private ProductService service;
    
    private RSocketRequester requester;
    
    @GetMapping("all")
    public Flux<ProductDto> all(){
        this.requester.route("post.findAll")
                .retrieveFlux(ProductDto.class);
                return this.service.getAll();
    }