如何使用reactor和R2dbc压缩嵌套列表

Question

我在postgres数据库中有3个表，并使用R2dbc以关系方式查询和连接它们。

我有3个实体类(可能不应该是数据类，但不应该影响示例(

@Entity
@Table(name = "parent", schema = "public", catalog = "Test")
data class MyParentObject(
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Id
@org.springframework.data.annotation.Id
@Column(name = "id")
var id: Int = 0,
@Transient
var childData: List<MyChildObject>? = null
)

@Entity
@Table(name = "child", schema = "public", catalog = "Test")
data class MyChildObject(
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Id
@org.springframework.data.annotation.Id
@Column(name = "id")
var id: Int = 0,
@Column(name = "parent_id")
var parentId: Int? = null
@Transient
var grandchildData: List<MyGrandchildObject>? = null
)

@Entity
@Table(name = "grandchild", schema = "public", catalog = "Test")
data class MyGrandchildObject(
@GeneratedValue(strategy = GenerationType.IDENTITY)
@Id
@org.springframework.data.annotation.Id
@Column(name = "id")
var id: Int = 0

@Column(name = "child_id")
var childId: Int? = null
)

父母是孩子的一对多，也就是孙子的一对很多。parent_id和child_id的行为就像fkeys。

我有一个RestController，它可以通过这些方法返回所有填充了所有子数据的父数据

fun viewAllParents(): Mono<MutableList<MyParentObject>> =
parentRepository.findAll()
.flatMap { Mono.just(it).addChildData(it.id) }
.collectList()
fun Mono<MyParentObject>.addChildData(id: Int): Mono<MyParentObject> =
this.zipWith(childRepository.getAllByParentIdEquals(id).collectList())
.map {
it.t1.childData = it.t2
it.t1
}

我有另一个RestController，它可以通过这些方法返回所有ChildData和所有孙子数据(与上面的非常相同(

fun viewAllChildren(): Mono<MutableList<MyChildObject>> =
childRepository.findAll()
.flatMap { Mono.just(it).addGrandchildData(it.id) }
.collectList()
fun Mono<MyChildObject>.addGrandchildData(id: Int): Mono<MyChildObject> =
this.zipWith(childOfChildRepository.getAllByChildIdEquals(id).collectList())
.map {
it.t1.childOfChildData = it.t2
it.t1
}

我不能做的，也是我的问题，是如何让viewAllParents()也填充孙子数据。我是否需要将var grandchildData: List<MyGrandchildObject>转换为Flux，并使用孙库中的新Flux进行压缩？还是我看错了方向？

任何建议都将不胜感激。

Answer 1

我真的很喜欢使用reactor进行分层数据提取的挑战。我不认识Kotlin，但我尝试过用java重现这个问题。我无法使用父级创建PostgreSQL表->child->grandChild层次结构，但我尝试通过web客户端模拟类似的东西(基本上逻辑保持不变(。这是我的代码，这是我试图做的，并且能够得到你想要的结果：https://github.com/harryalto/reactive-springwebflux

解决方案的关键是在Handler代码中，我使用它来构建基于子级列表的子流，并使用它将所有连接在一起

public Flux<Parent> getFamiliesHierarchy() {
return getAllParents()
.flatMap(parent ->
getAllChildsList(parent.getId())
.flatMap(childList -> getChildsWithKids(childList))
.map(childList -> parent.toBuilder().children(childList).build()
)
);
}

以下是完整的代码

@Component
@Slf4j
public class FamilyHandler {
@Autowired
private WebClient webClient;
public Flux<Parent> getAllParents() {
return webClient
.get()
.uri("parents")
.retrieve()
.bodyToFlux(Parent.class);
}
public Mono<List<Child>> getAllChildsList(final Integer parentId) {
ParameterizedTypeReference<List<Child>> childList = 
new ParameterizedTypeReference<List<Child>>() {};
return webClient
.get()
.uri("childs?parentId=" + parentId)
.retrieve()
.bodyToMono(childList);
}
public Flux<GrandChild> getGrandKids(final Integer childId) {
return webClient
.get()
.uri("grandKids?childId=" + childId)
.retrieve()
.bodyToFlux(GrandChild.class);
}
public Mono<List<GrandChild>> getGrandKidsList(final Integer childId) {
ParameterizedTypeReference<List<GrandChild>> grandKidsList = 
new ParameterizedTypeReference<List<GrandChild>>() {};
return webClient
.get()
.uri("grandKids?childId=" + childId)
.retrieve()
.bodyToMono(grandKidsList);
}
private Mono<List<Child>> getChildsWithKids(final List<Child> childList) {
return Flux.fromIterable(childList).flatMap(child ->
Mono.zip(Mono.just(child), getGrandKidsList(child.getId()))
.map(tuple2 ->        tuple2.getT1().toBuilder().grandChildren(tuple2.getT2()).build())
).collectList();
}
public Flux<Parent> getFamiliesHierarchy() {
return getAllParents()
.flatMap(parent ->
getAllChildsList(parent.getId())
.flatMap(childList -> getChildsWithKids(childList))
.map(childList -> parent.toBuilder().children(childList).build()
)
);
}
}`

我使用json服务器来嘲笑服务器

下面是我的db.json文件

{
"parents":[
{
"id": 1,
"name" : "Parent1",
"path":"1"
},
{
"id": 2,
"name" : "Parent2",
"path":"2"
}
],
"childs":[
{
"id": 1,
"parentId": 1,
"name": "child1Parent1",
"path":"1.1"
},
{
"id":2,
"parentId": 1,
"projectName": "child2Parent1",
"path":"1.2"
},
{
"id":3,
"parentId": 2,
"projectName": "child1Parent2",
"path":"2.1"
},
{
"id":4,
"parentId": 2,
"projectName": "child2Parent2",
"path":"2.2"
}
],
"grandKids":[
{
"id":1,
"childId": 2,
"projectName": "grandKid1child2Parent1",
"path":"1.2.1"
},
{
"id":3,
"childId": 2,
"projectName": "grandKid1child2Parent1",
"path":"1.2.2"
},
{
"id":2,
"childId": 4,
"projectName": "grandKid1child1Parent2",
"path":"2.2.1"
},
{
"id":4,
"childId": 4,
"projectName": "grandKid1child1Parent2",
"path":"2.2.2"
},
{
"id":5,
"childId": 3,
"projectName": "grandKid1child1Parent2",
"path":"2.1.1"
}

]
}

这是我的控制器代码

@RestController
@Slf4j
public class FamilyController {
@Autowired
private FamilyHandler familyHandler;
@GetMapping(FAMILIES_ENDPOINT_V1)
public Flux<Parent> viewAllParents() {
return familyHandler.getFamiliesHierarchy();
}
}

我们可以很容易地为r2DBC存储库修改代码。

--更新--我能够创建示例数据，并使用R2DBC创建了一个等效数据这是要点的链接