/**
 * Pull the data stored in Redis at the given keys as bytes using the mget command. If no data is
 * stored at a given key in Redis, will subsitute the data with null.
 *
 * @param keys list of {@link RedisKey} to pull from redis.
 * @return list of data bytes or null pulled from redis for each given key.
 */
private List<byte[]> sendMultiGet(List<RedisKey> keys) {
  try {
    byte[][] binaryKeys =
        keys.stream()
            .map(AbstractMessageLite::toByteArray)
            .collect(Collectors.toList())
            .toArray(new byte[0][0]);
    return syncCommands.mget(binaryKeys).stream()
        .map(
            keyValue -> {
              if (keyValue == null) {
                return null;
              }
              return keyValue.getValueOrElse(null);
            })
        .collect(Collectors.toList());
  } catch (Exception e) {
    throw Status.NOT_FOUND
        .withDescription("Unable to retrieve feature from Redis")
        .withCause(e)
        .asRuntimeException();
  }
}
 

/**
 * Pull the data stored in Redis at the given keys as bytes using the mget command. If no data is
 * stored at a given key in Redis, will subsitute the data with null.
 *
 * @param keys list of {@link RedisKey} to pull from redis.
 * @return list of data bytes or null pulled from redis for each given key.
 */
private List<byte[]> sendMultiGet(List<RedisKey> keys) {
  try {
    byte[][] binaryKeys =
        keys.stream()
            .map(AbstractMessageLite::toByteArray)
            .collect(Collectors.toList())
            .toArray(new byte[0][0]);
    return syncCommands.mget(binaryKeys).stream()
        .map(
            keyValue -> {
              if (keyValue == null) {
                return null;
              }
              return keyValue.getValueOrElse(null);
            })
        .collect(Collectors.toList());
  } catch (Exception e) {
    throw Status.UNKNOWN
        .withDescription("Unexpected error when pulling data from from Redis.")
        .withCause(e)
        .asRuntimeException();
  }
}
 

@Before
public void setUp() {
  initMocks(this);
  when(connection.sync()).thenReturn(syncCommands);
  redisClusterOnlineRetriever = RedisClusterOnlineRetriever.create(connection);
  redisKeyList =
      Lists.newArrayList(
              RedisKey.newBuilder()
                  .setFeatureSet("project/featureSet")
                  .addAllEntities(
                      Lists.newArrayList(
                          Field.newBuilder().setName("entity1").setValue(intValue(1)).build(),
                          Field.newBuilder().setName("entity2").setValue(strValue("a")).build()))
                  .build(),
              RedisKey.newBuilder()
                  .setFeatureSet("project/featureSet")
                  .addAllEntities(
                      Lists.newArrayList(
                          Field.newBuilder().setName("entity1").setValue(intValue(2)).build(),
                          Field.newBuilder().setName("entity2").setValue(strValue("b")).build()))
                  .build())
          .stream()
          .map(AbstractMessageLite::toByteArray)
          .collect(Collectors.toList())
          .toArray(new byte[0][0]);
}
 

@Before
public void setUp() {
  initMocks(this);
  when(connection.sync()).thenReturn(syncCommands);
  redisOnlineRetriever = RedisOnlineRetriever.create(connection);
  redisKeyList =
      Lists.newArrayList(
              RedisKey.newBuilder()
                  .setFeatureSet("project/featureSet")
                  .addAllEntities(
                      Lists.newArrayList(
                          Field.newBuilder().setName("entity1").setValue(intValue(1)).build(),
                          Field.newBuilder().setName("entity2").setValue(strValue("a")).build()))
                  .build(),
              RedisKey.newBuilder()
                  .setFeatureSet("project/featureSet")
                  .addAllEntities(
                      Lists.newArrayList(
                          Field.newBuilder().setName("entity1").setValue(intValue(2)).build(),
                          Field.newBuilder().setName("entity2").setValue(strValue("b")).build()))
                  .build())
          .stream()
          .map(AbstractMessageLite::toByteArray)
          .collect(Collectors.toList())
          .toArray(new byte[0][0]);
}
 

@Test
public void readEvents_multipleEventsInOneChunk() throws Exception {
    final List<Event> subHbOffer = newArrayList(
        TestingProtos.SUBSCRIBED,
        TestingProtos.HEARTBEAT,
        TestingProtos.OFFER
    );
    final List<byte[]> eventChunks = subHbOffer.stream()
        .map(AbstractMessageLite::toByteArray)
        .map(RecordIOUtils::createChunk)
        .collect(Collectors.toList());
    final List<ByteBuf> singleChunk = newArrayList(Unpooled.copiedBuffer(concatAllChunks(eventChunks)));

    final List<Event> events = runTestOnChunks(singleChunk);
    assertThat(events).isEqualTo(subHbOffer);
}
 

/**
 * Write the proto to a file in the app' s file directory.
 *
 * <p>Writes are non atomic.
 *
 * <p>Readers are expected to deal with corrupt data resulting from faulty writes
 *
 * @param messageLite
 * @throws IOException
 */
public Completable write(AbstractMessageLite messageLite) {
  return Completable.fromCallable(
      () -> {
        // reads / writes are synchronized per client instance
        synchronized (this) {
          try (FileOutputStream output = application.openFileOutput(fileName, MODE_PRIVATE)) {
            output.write(messageLite.toByteArray());
            return messageLite;
          }
        }
      });
}
 

public static String encode(AbstractMessageLite message) {
  return new String(Base64.encodeBase64(message.toByteArray()), CHAR_SET);
}
 

/** Constructs a {@link ProtoDeterministicWriter} with an eagerly constructed message. */
public ProtoDeterministicWriter(AbstractMessageLite<?, ?> message) {
  this.messageSupplier = () -> message;
}
 

/** Returns a ProtoWrapper that wraps the provided object */
public static <M extends AbstractMessageLite> ProtoWrapper<M> of(M proto) {
  return new ProtoWrapper<M>(proto);
}
 

/** Returns a ProtoWrapper that wraps the provided object */
public static <M extends AbstractMessageLite> ProtoWrapper<M> of(M proto) {
  return new ProtoWrapper<M>(proto);
}
 

/**
 * Read the contents of the file into a proto object using the parser. Since writes are not
 * atomic, the caller will receive {@link Maybe#empty()} when data is corrupt.
 *
 * <p>Some valid scenarios that can lead to corrupt data :
 *
 * <ul>
 *   <li>Out of disk space while writing
 *   <li>Power outage while writing
 *   <li>Process killed while writing
 * </ul>
 *
 * @param parser
 * @param <T>
 */
public <T extends AbstractMessageLite> Maybe<T> read(Parser<T> parser) {
  return Maybe.fromCallable(
      () -> {
        // reads / writes are synchronized per client instance
        synchronized (this) {
          try (FileInputStream inputStream = application.openFileInput(fileName)) {
            return parser.parseFrom(inputStream);
          } catch (InvalidProtocolBufferException | FileNotFoundException e) {
            Logging.logi("Recoverable exception while reading cache: " + e.getMessage());
            return null;
          }
        }
      });
}
 

AbstractMessageLite<?, ?> getMessage() throws IOException;