juplo.de Git - demos/kafka/chat/blob - src/main/java/de/juplo/kafka/chat/backend/storage/files/FilesStorageStrategy.java

   1 package de.juplo.kafka.chat.backend.storage.files;
   2
   3 import com.fasterxml.jackson.core.JsonGenerator;
   4 import com.fasterxml.jackson.databind.JavaType;
   5 import com.fasterxml.jackson.databind.ObjectMapper;
   6 import de.juplo.kafka.chat.backend.api.ChatRoomInfoTo;
   7 import de.juplo.kafka.chat.backend.api.MessageTo;
   8 import de.juplo.kafka.chat.backend.domain.ChatRoomInfo;
   9 import de.juplo.kafka.chat.backend.domain.Message;
  10 import de.juplo.kafka.chat.backend.implementation.StorageStrategy;
  11 import de.juplo.kafka.chat.backend.implementation.ShardingStrategy;
  12 import lombok.RequiredArgsConstructor;
  13 import lombok.extern.slf4j.Slf4j;
  14 import reactor.core.publisher.Flux;
  15
  16 import java.io.IOException;
  17 import java.nio.file.Files;
  18 import java.nio.file.Path;
  19 import java.util.UUID;
  20
  21 import static java.nio.file.StandardOpenOption.CREATE;
  22 import static java.nio.file.StandardOpenOption.TRUNCATE_EXISTING;
  23
  24
  25 @RequiredArgsConstructor
  26 @Slf4j
  27 public class FilesStorageStrategy implements StorageStrategy
  28 {
  29   public static final String CHATROOMS_FILENAME = "chatrooms.json";
  30
  31
  32   private final Path storagePath;
  33   private final ShardingStrategy shardingStrategy;
  34   private final ObjectMapper mapper;
  35
  36
  37   @Override
  38   public void writeChatRoomInfo(Flux<ChatRoomInfo> chatRoomInfoFlux)
  39   {
  40     Path path = chatroomsPath();
  41     log.info("Writing chatrooms to {}", path);
  42     try
  43     {
  44       Files.createDirectories(storagePath);
  45
  46       JsonGenerator generator =
  47           mapper
  48               .getFactory()
  49               .createGenerator(Files.newBufferedWriter(path, CREATE, TRUNCATE_EXISTING));
  50
  51       chatRoomInfoFlux
  52           .log()
  53           .doFirst(() ->
  54           {
  55             try
  56             {
  57               generator.useDefaultPrettyPrinter();
  58               generator.writeStartArray();
  59             }
  60             catch (IOException e)
  61             {
  62               throw new RuntimeException(e);
  63             }
  64           })
  65           .doOnNext(chatRoomInfo ->
  66           {
  67             try
  68             {
  69               ChatRoomInfoTo chatRoomInfoTo = ChatRoomInfoTo.from(chatRoomInfo);
  70               generator.writeObject(chatRoomInfoTo);
  71             }
  72             catch (IOException e)
  73             {
  74               throw new RuntimeException(e);
  75             }
  76           })
  77           .doOnTerminate(() ->
  78           {
  79             try
  80             {
  81               generator.writeEndArray();
  82               generator.close();
  83             }
  84             catch (IOException e)
  85             {
  86               throw new RuntimeException(e);
  87             }
  88           });
  89     }
  90     catch (IOException e)
  91     {
  92       throw new RuntimeException(e);
  93     }
  94   }
  95
  96   @Override
  97   public Flux<ChatRoomInfo> readChatRoomInfo()
  98   {
  99     JavaType type = mapper.getTypeFactory().constructType(ChatRoomInfoTo.class);
 100     return Flux
 101         .from(new JsonFilePublisher<ChatRoomInfoTo>(chatroomsPath(), mapper, type))
 102         .log()
 103         .map(chatRoomInfoTo ->
 104         {
 105           UUID chatRoomId = chatRoomInfoTo.getId();
 106           int shard = shardingStrategy.selectShard(chatRoomId);
 107
 108           log.info(
 109               "{} - old shard: {}, new shard:  {}",
 110               chatRoomId,
 111               chatRoomInfoTo.getShard(),
 112               shard);
 113
 114           return new ChatRoomInfo(
 115               chatRoomId,
 116               chatRoomInfoTo.getName(),
 117               shard);
 118         });
 119   }
 120
 121   @Override
 122   public void writeChatRoomData(
 123       UUID chatRoomId,
 124       Flux<Message> messageFlux,
 125       SuccessCallback successCallback,
 126       FailureCallback failureCallback)
 127   {
 128     Path path = chatroomPath(chatRoomId);
 129     log.info("Writing messages for {} to {}", chatRoomId, path);
 130     try
 131     {
 132       Files.createDirectories(storagePath);
 133
 134       JsonGenerator generator =
 135           mapper
 136               .getFactory()
 137               .createGenerator(Files.newBufferedWriter(path, CREATE, TRUNCATE_EXISTING));
 138
 139       messageFlux
 140           .log()
 141           .doFirst(() ->
 142           {
 143             try
 144             {
 145               generator.useDefaultPrettyPrinter();
 146               generator.writeStartArray();
 147             }
 148             catch (IOException e)
 149             {
 150               throw new RuntimeException(e);
 151             }
 152           })
 153           .doOnTerminate(() ->
 154           {
 155             try
 156             {
 157               generator.writeEndArray();
 158               generator.close();
 159             }
 160             catch (IOException e)
 161             {
 162               throw new RuntimeException(e);
 163             }
 164           })
 165           .subscribe(message ->
 166           {
 167             try
 168             {
 169               MessageTo messageTo = MessageTo.from(message);
 170               generator.writeObject(messageTo);
 171             }
 172             catch (IOException e)
 173             {
 174               throw new RuntimeException(e);
 175             }
 176           });
 177
 178       successCallback.accept(chatRoomId);
 179     }
 180     catch (IOException e)
 181     {
 182       failureCallback.accept(chatRoomId, e);
 183     }
 184   }
 185
 186   @Override
 187   public Flux<Message> readChatRoomData(UUID chatRoomId)
 188   {
 189     JavaType type = mapper.getTypeFactory().constructType(MessageTo.class);
 190     return Flux
 191         .from(new JsonFilePublisher<MessageTo>(chatroomPath(chatRoomId), mapper, type))
 192         .log()
 193         .map(MessageTo::toMessage);
 194   }
 195
 196   Path chatroomsPath()
 197   {
 198     return storagePath.resolve(Path.of(CHATROOMS_FILENAME));
 199   }
 200
 201   Path chatroomPath(UUID id)
 202   {
 203     return storagePath.resolve(Path.of(id.toString() + ".json"));
 204   }
 205 }