[Day 26] Reactive Programming - Spring WebFlux(R2DBC Repositories)

前言

上一篇我们使用ReactiveCrudRepository来对资料库存取，对於一些不太复杂的SQL指令来说，使用CrudRepository方便又省事，让我们来看看在Reactive的世界中，是否仍然始终如一呢？

R2DBC Repositories

Spring提供另外一个R2dbcRepository，ReactiveSortingRepository继承ReactiveCrudRepository，所以整体来说R2dbcRepository的功能会更加的全面。

public interface R2dbcRepository<T, ID> extends ReactiveSortingRepository<T, ID>, ReactiveQueryByExampleExecutor<T> {}

Query Methods

从上一篇的范例稍微调整一下(先替换为R2dbcRepository)，与Spring Data最明显的差异可想而知就是回传的型别，

1. 原本回传单笔的型态Optional<>改为Mono<>，多笔则换成了Flux<>。
2. 另外则是多了可以传入Publisher的方法，可以用stream的方式一个一个传入message来查询。
3. 再来就是关於分页(page)，虽然有提供Pageable 可以达到分页的效果，但回传的类别仍然是Flux<>而不是原本的Page，当然是可以额外自己count，推测Spring Data R2DBC没有特别处理这个的原因或许是分页某种程度上就是避免大量资料造成效能相关问题，但Reactive本身就有一些方式能解决这样的问题。

public interface GreetingRepository extends R2dbcRepository<Greeting, Long> {
  Mono<Greeting> findById(Long id);

  Flux<Greeting> findAll();
  Flux<Greeting> findByMessage(String message, Pageable pageable);
  Flux<Greeting> findByMessage((Publisher<String> message);

  Mono<Void> save(Mono<Greeting> greeting);
}

详细可参考doc

Modifying Queries

针对修改类型的有三种回传的型态

1. Interger：更改的笔数
2. Boolean:  是否有任一笔更改
3. Void：不在意执行结果，没有任何回传。
   若是使用自订的Query来做update则需要加上@Modifying的annotation，原本的Spring data一样也需要。

Mono<Integer> deleteByMessage(String message);
//Mono<Void> deleteByMessage(String message);
//Mono<Boolean> deleteByMessage(String message);
@Modifying
@Query("UPDATE message SET message = :message where id = :id")
Mono<Integer> updateMessage(String message, Long id);

ID

仍有支援ID Generation，上一个范例就是透过mySql 自动产生流水号ID。判断entity是否已存在DB的方式与Spring Data相同，常见的方式如下：

1. @ID：被挂上@ID的栏位是否为空，若为null则该entity会被视为一个新的entity，就会使用insert的语法反之则会用update。
2. Version：被挂上@Version如果是0或是null，则被视为null。
3. 实作Persistable介面：自行实作isNew()，Spring data会透过isNew()来判断。

最後根据issue目前还不支援组合键(compositeId)。

convert

或多或少都会有需求要客制物件，现在假设原本的Greeting需要多两个栏位，一个单纯的Y、N，另一个则是在DB里面希望显示数字，而程序则需显示英文，先不讨论需求合理性与有没有别的作法，单纯就是DEMO一次客制转换的方式。
首先新增两个Enum，在这边预设甚麽都没加，enum会转换成string，这在以前记得Enum预设是转换为enum的数字(顺序)而不是文字。另一个则是希望DB储存是自订栏位而不是文字，所以增加了一个typeCode的属性。

public enum YesNo { 
  Y, 
  N 
  ; 
}

@Getter 
public enum MessageType { 
  VOICE("1"), 
  MP3("2"), 
  WORD("3") 
  ; 
  private String typeCode; 
  MessageType(String typeCode) { 
    this.typeCode = typeCode; 
  } 
  private static final Map<String, MessageType> BY_CODE = new HashMap<>(MessageType.values().length); 
  static { 
    for (MessageType e : MessageType.values()) { 
      BY_CODE.put(e.getTypeCode(), e); 
    } 
  } 
  public static MessageType getTypeFromCode(String typeCode) { 
    return BY_CODE.getOrDefault(typeCode, WORD); 
  } 
}

public class Greeting {
  @Id
  private Long id;
  private String message;
  private YesNo isSend;
  private MessageType messageType;
  
  public Greeting(String message) {
    this.message = message;
  }
  public Greeting(Long id, String message) {
    this.id = id;
    this.message = message;
  }

  public Greeting(String message, YesNo isSend,
      MessageType messageType) {
    this.message = message;
    this.isSend = isSend;
    this.messageType = messageType;
  }
}

现在就需要特别设定自定义的Converter以及注册到R2dbcConfiguration中，分别有

1. AppConfig：注册。
2. MessageTypeReadConverter：从db转回物件，要注意converter是springframework.core里面的。
3. MessageTypeWriteConverter：从物件写入db。

@ReadingConverter
public class MessageTypeReadConverter
    implements org.springframework.core.convert.converter.Converter<String, MessageType> {

  @Override
  public MessageType convert(String source) {
    return MessageType.getTypeFromCode(source);
  }
}

@WritingConverter
public class MessageTypeWriteConverter implements Converter<MessageType, String> {

  @Override
  public String convert(MessageType source) {
    return source.getTypeCode();
  }
}

@Configuration
public class AppConfig extends AbstractR2dbcConfiguration {

  @Override
  public ConnectionFactory connectionFactory() {
    return ConnectionFactories.get("r2dbc:mysql://localhost:3306/test");
  }

  @Override
  protected List<Object> getCustomConverters() {

    return List.of(
        new MessageTypeReadConverter(),
        new MessageTypeWriteConverter()
    );
  }
}

测试後成功将资料写入DB时会进行转换，附上实际DB资料。

结语

这次的实作感想是觉得Spring对於RDB的支援度还不够全面，对於NoSql比较友善，像是这次使用mySql的R2DBC Driver  0.8.2，R2DBC官方DOC，是用OutboundRow来做转换，但mySql R2DBC Driver的codec并不认识OutboundRow还需要自己再转换一次，所以上面的范例索性就直接用String而不是OutboundRow，但这样是如果情境更复杂需要多栏位组合就没有办法。

@WritingConverter 
public class PersonWriteConverter implements Converter<Person, OutboundRow> { 
  public OutboundRow convert(Person source) { 
    OutboundRow row = new OutboundRow(); 
    row.put("id", SettableValue.from(source.getId())); 
    row.put("name", SettableValue.from(source.getFirstName())); 
    row.put("age", SettableValue.from(source.getAge())); 
    return row; 
  } 
}