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           .doOnTerminate(() ->
  66           {
  67             try
  68             {
  69               generator.writeEndArray();
  70               generator.close();
  71             }
  72             catch (IOException e)
  73             {
  74               throw new RuntimeException(e);
  75             }
  76           })
  77           .map(chatRoomInfo ->
  78           {
  79             try
  80             {
  81               ChatRoomInfoTo chatRoomInfoTo = ChatRoomInfoTo.from(chatRoomInfo);
  82               generator.writeObject(chatRoomInfoTo);
  83               return chatRoomInfo;
  84             }
  85             catch (IOException e)
  86             {
  87               throw new RuntimeException(e);
  88             }
  89           })
  90           .subscribe();
  91     }
  92     catch (IOException e)
  93     {
  94       throw new RuntimeException(e);
  95     }
  96   }
  97
  98   @Override
  99   public Flux<ChatRoomInfo> readChatRoomInfo()
 100   {
 101     JavaType type = mapper.getTypeFactory().constructType(ChatRoomInfoTo.class);
 102     return Flux
 103         .from(new JsonFilePublisher<ChatRoomInfoTo>(chatroomsPath(), mapper, type))
 104         .log()
 105         .map(chatRoomInfoTo ->
 106         {
 107           UUID chatRoomId = chatRoomInfoTo.getId();
 108           int shard = shardingStrategy.selectShard(chatRoomId);
 109
 110           log.info(
 111               "{} - old shard: {}, new shard:  {}",
 112               chatRoomId,
 113               chatRoomInfoTo.getShard(),
 114               shard);
 115
 116           return new ChatRoomInfo(
 117               chatRoomId,
 118               chatRoomInfoTo.getName(),
 119               shard);
 120         });
 121   }
 122
 123   @Override
 124   public void writeChatRoomData(
 125       UUID chatRoomId,
 126       Flux<Message> messageFlux)
 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           .map(message ->
 166           {
 167             try
 168             {
 169               MessageTo messageTo = MessageTo.from(message);
 170               generator.writeObject(messageTo);
 171               return message;
 172             }
 173             catch (IOException e)
 174             {
 175               throw new RuntimeException(e);
 176             }
 177           })
 178           .subscribe();
 179     }
 180     catch (IOException e)
 181     {
 182       throw new RuntimeException(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 }