@Test public void testStoreEmptyUsername() throws Exception { try { store.store("", CURSOR_NAME, "1"); failBecauseExceptionWasNotThrown(IllegalArgumentException.class); } catch (IllegalArgumentException e) { } verify(client, never()).execute(any(UpdateValue.class)); }
public void store(final String username, final String cursorName, final String value) throws NotificationStoreException { Objects.requireNonNull(username, "username == null"); Preconditions.checkArgument(!username.isEmpty(), "username cannot be empty"); Objects.requireNonNull(cursorName, "cursorName == null"); Preconditions.checkArgument(!cursorName.isEmpty(), "cursorName cannot be empty"); final String key = getCursorKey(username, cursorName); final CursorUpdate update = new CursorUpdate(key, value); final Location location = new Location(NAMESPACE, key); final UpdateValue updateValue = new UpdateValue.Builder(location) .withUpdate(update) .withStoreOption(StoreValue.Option.RETURN_BODY, false) .withTimeout(timeout) .build(); LOGGER.debug("Updating cursor ({}) to value (async): {}", location, value); try (Timer.Context context = storeTimer.time()) { final RiakFuture<UpdateValue.Response, Location> future = client.executeAsync(updateValue); future.await(requestTimeout.getQuantity(), requestTimeout.getUnit()); if (future.isSuccess()) { LOGGER.debug("Successfully updated cursor: {}", location); } } catch (InterruptedException e) { LOGGER.warn("Update request was interrupted", e); Thread.currentThread().interrupt(); throw new NotificationStoreException(e); } }
CursorStore { public void store(final String username, final String cursorName, final String value) throws NotificationStoreException { Objects.requireNonNull(username, "username == null"); Preconditions.checkArgument(!username.isEmpty(), "username cannot be empty"); Objects.requireNonNull(cursorName, "cursorName == null"); Preconditions.checkArgument(!cursorName.isEmpty(), "cursorName cannot be empty"); final String key = getCursorKey(username, cursorName); final CursorUpdate update = new CursorUpdate(key, value); final Location location = new Location(NAMESPACE, key); final UpdateValue updateValue = new UpdateValue.Builder(location) .withUpdate(update) .withStoreOption(StoreValue.Option.RETURN_BODY, false) .withTimeout(timeout) .build(); LOGGER.debug("Updating cursor ({}) to value (async): {}", location, value); try (Timer.Context context = storeTimer.time()) { final RiakFuture<UpdateValue.Response, Location> future = client.executeAsync(updateValue); future.await(requestTimeout.getQuantity(), requestTimeout.getUnit()); if (future.isSuccess()) { LOGGER.debug("Successfully updated cursor: {}", location); } } catch (InterruptedException e) { LOGGER.warn("Update request was interrupted", e); Thread.currentThread().interrupt(); throw new NotificationStoreException(e); } } }
CursorStore { public void store(final String username, final String cursorName, final String value) throws NotificationStoreException { Objects.requireNonNull(username, "username == null"); Preconditions.checkArgument(!username.isEmpty(), "username cannot be empty"); Objects.requireNonNull(cursorName, "cursorName == null"); Preconditions.checkArgument(!cursorName.isEmpty(), "cursorName cannot be empty"); final String key = getCursorKey(username, cursorName); final CursorUpdate update = new CursorUpdate(key, value); final Location location = new Location(NAMESPACE, key); final UpdateValue updateValue = new UpdateValue.Builder(location) .withUpdate(update) .withStoreOption(StoreValue.Option.RETURN_BODY, false) .withTimeout(timeout) .build(); LOGGER.debug("Updating cursor ({}) to value (async): {}", location, value); try (Timer.Context context = storeTimer.time()) { final RiakFuture<UpdateValue.Response, Location> future = client.executeAsync(updateValue); future.await(requestTimeout.getQuantity(), requestTimeout.getUnit()); if (future.isSuccess()) { LOGGER.debug("Successfully updated cursor: {}", location); } } catch (InterruptedException e) { LOGGER.warn("Update request was interrupted", e); Thread.currentThread().interrupt(); throw new NotificationStoreException(e); } } CursorStore(
      final RiakClient client, final Duration timeout, final Duration requestTimeout); }
CursorStore { public void store(final String username, final String cursorName, final String value) throws NotificationStoreException { Objects.requireNonNull(username, "username == null"); Preconditions.checkArgument(!username.isEmpty(), "username cannot be empty"); Objects.requireNonNull(cursorName, "cursorName == null"); Preconditions.checkArgument(!cursorName.isEmpty(), "cursorName cannot be empty"); final String key = getCursorKey(username, cursorName); final CursorUpdate update = new CursorUpdate(key, value); final Location location = new Location(NAMESPACE, key); final UpdateValue updateValue = new UpdateValue.Builder(location) .withUpdate(update) .withStoreOption(StoreValue.Option.RETURN_BODY, false) .withTimeout(timeout) .build(); LOGGER.debug("Updating cursor ({}) to value (async): {}", location, value); try (Timer.Context context = storeTimer.time()) { final RiakFuture<UpdateValue.Response, Location> future = client.executeAsync(updateValue); future.await(requestTimeout.getQuantity(), requestTimeout.getUnit()); if (future.isSuccess()) { LOGGER.debug("Successfully updated cursor: {}", location); } } catch (InterruptedException e) { LOGGER.warn("Update request was interrupted", e); Thread.currentThread().interrupt(); throw new NotificationStoreException(e); } } CursorStore(
      final RiakClient client, final Duration timeout, final Duration requestTimeout); void initialize(); Optional<String> fetch(final String username, final String cursorName); void store(final String username, final String cursorName, final String value); void delete(final String username, final String cursorName); String getCursorKey(final String username, final String cursorName); }
CursorStore { public void store(final String username, final String cursorName, final String value) throws NotificationStoreException { Objects.requireNonNull(username, "username == null"); Preconditions.checkArgument(!username.isEmpty(), "username cannot be empty"); Objects.requireNonNull(cursorName, "cursorName == null"); Preconditions.checkArgument(!cursorName.isEmpty(), "cursorName cannot be empty"); final String key = getCursorKey(username, cursorName); final CursorUpdate update = new CursorUpdate(key, value); final Location location = new Location(NAMESPACE, key); final UpdateValue updateValue = new UpdateValue.Builder(location) .withUpdate(update) .withStoreOption(StoreValue.Option.RETURN_BODY, false) .withTimeout(timeout) .build(); LOGGER.debug("Updating cursor ({}) to value (async): {}", location, value); try (Timer.Context context = storeTimer.time()) { final RiakFuture<UpdateValue.Response, Location> future = client.executeAsync(updateValue); future.await(requestTimeout.getQuantity(), requestTimeout.getUnit()); if (future.isSuccess()) { LOGGER.debug("Successfully updated cursor: {}", location); } } catch (InterruptedException e) { LOGGER.warn("Update request was interrupted", e); Thread.currentThread().interrupt(); throw new NotificationStoreException(e); } } CursorStore(
      final RiakClient client, final Duration timeout, final Duration requestTimeout); void initialize(); Optional<String> fetch(final String username, final String cursorName); void store(final String username, final String cursorName, final String value); void delete(final String username, final String cursorName); String getCursorKey(final String username, final String cursorName);  }