DATAREDIS-697 - Add support for BITPOS.

We now support the BITPOS command throughout the sync and reactive api.

Original pull request: #335.

Author: christophstrobl

src/main/java/org/springframework/data/redis/connection/DefaultStringRedisConnection.java

@@ -1160,6 +1160,16 @@ public Long bitOp(BitOperation op, byte[] destination, byte[]... keys) {
 		return convertAndReturn(delegate.bitOp(op, destination, keys), identityConverter);
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#bitPos(byte[], boolean, org.springframework.data.domain.Range)
+	 */
+	@Nullable
+	@Override
+	public Long bitPos(byte[] key, boolean bit, org.springframework.data.domain.Range<Long> range) {
+ 		return convertAndReturn(delegate.bitPos(key, bit, range), identityConverter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisPubSubCommands#subscribe(org.springframework.data.redis.connection.MessageListener, byte[][])
@@ -2470,6 +2480,16 @@ public Long bitOp(BitOperation op, String destination, String... keys) {
 		return bitOp(op, serialize(destination), serializeMulti(keys));
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#bitPos(java.lang.String, boolean, org.springframework.data.domain.Range)
+	 */
+	@Nullable
+	@Override
+	public Long bitPos(String key, boolean bit, org.springframework.data.domain.Range<Long> range) {
+		return bitPos(serialize(key), bit, range);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.StringRedisConnection#subscribe(org.springframework.data.redis.connection.MessageListener, java.lang.String[])

src/main/java/org/springframework/data/redis/connection/DefaultedRedisConnection.java

@@ -22,6 +22,7 @@
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.data.domain.Range;
 import org.springframework.data.geo.Circle;
 import org.springframework.data.geo.Distance;
 import org.springframework.data.geo.GeoResults;
@@ -385,6 +386,13 @@ default Long bitOp(BitOperation op, byte[] destination, byte[]... keys) {
 		return stringCommands().bitOp(op, destination, keys);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
+	@Override
+	@Deprecated
+	default Long bitPos(byte[] key, boolean bit, org.springframework.data.domain.Range<Long> range) {
+		return stringCommands().bitPos(key, bit, range);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
 	@Override
 	@Deprecated

src/main/java/org/springframework/data/redis/connection/ReactiveStringCommands.java

@@ -1014,6 +1014,80 @@ default Mono<Long> bitOp(Collection<ByteBuffer> keys, BitOperation bitOp, ByteBu
 	 */
 	Flux<NumericResponse<BitOpCommand, Long>> bitOp(Publisher<BitOpCommand> commands);
 
+	/**
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	class BitPosCommand extends KeyCommand {
+
+		private boolean bit;
+		private Range<Long> range;
+
+		private BitPosCommand(@Nullable ByteBuffer key, boolean bit, Range<Long> range) {
+			super(key);
+			this.bit = bit;
+			this.range = range;
+		}
+
+		static BitPosCommand positionOf(boolean bit) {
+			return new BitPosCommand(null, bit, Range.unbounded());
+		}
+
+		public BitPosCommand in(ByteBuffer key) {
+			return new BitPosCommand(key, bit, range);
+		}
+
+		public BitPosCommand within(Range<Long> range) {
+			return new BitPosCommand(getKey(), bit, range);
+		}
+
+		public boolean getBit() {
+			return bit;
+		}
+
+		public Range<Long> getRange() {
+			return range;
+		}
+	}
+
+	/**
+	 * Return the position of the first bit set to given {@code bit} in a string.
+	 *
+	 * @param key the key holding the actual String.
+	 * @param bit the bit value to look for.
+	 * @return {@link Mono} emitting result when ready.
+	 * @since 2.1
+	 */
+	default Mono<Long> bitPos(ByteBuffer key, boolean bit) {
+		return bitPos(key, bit, Range.unbounded());
+	}
+
+	/**
+	 * Return the position of the first bit set to given {@code bit} in a string. {@link Range} start and end can contain
+	 * negative values in order to index <strong>bytes</strong> starting from the end of the string, where {@literal -1}
+	 * is the last byte, {@literal -2} is the penultimate.
+	 *
+	 * @param key the key holding the actual String.
+	 * @param bit the bit value to look for.
+	 * @param range must not be {@literal null}. Use {@link Range#unbounded()} to not limit search.
+	 * @return {@link Mono} emitting result when ready.
+	 * @since 2.1
+	 */
+	default Mono<Long> bitPos(ByteBuffer key, boolean bit, Range<Long> range) {
+		return bitPos(Mono.just(BitPosCommand.positionOf(bit).in(key).within(range))).next()
+				.map(NumericResponse::getOutput);
+	}
+
+	/**
+	 * Emmit the the position of the first bit set to given {@code bit} in a string. Get the length of the value stored at
+	 * {@literal key}.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return {@link Flux} emitting results when ready.
+	 * @since 2.1
+	 */
+	Flux<NumericResponse<BitPosCommand, Long>> bitPos(Publisher<BitPosCommand> commands);
+
 	/**
 	 * Get the length of the value stored at {@literal key}.
 	 *

src/main/java/org/springframework/data/redis/connection/RedisStringCommands.java

@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.Map;
 
+import org.springframework.data.domain.Range;
 import org.springframework.data.redis.core.types.Expiration;
 import org.springframework.lang.Nullable;
 
@@ -292,6 +293,37 @@ enum BitOperation {
 	@Nullable
 	Long bitOp(BitOperation op, byte[] destination, byte[]... keys);
 
+	/**
+	 * Return the position of the first bit set to given {@code bit} in a string.
+	 *
+	 * @param key the key holding the actual String.
+	 * @param bit the bit value to look for.
+	 * @return {@literal null} when used in pipeline / transaction. The position of the first bit set to 1 or 0 according
+	 *         to the request.
+	 * @see <a href="http://redis.io/commands/bitpos">Redis Documentation: BITPOS</a>
+	 * @since 2.1
+	 */
+	@Nullable
+	default Long bitPos(byte[] key, boolean bit) {
+		return bitPos(key, bit, Range.unbounded());
+	}
+
+	/**
+	 * Return the position of the first bit set to given {@code bit} in a string. {@link Range} start and end can contain
+	 * negative values in order to index <strong>bytes</strong> starting from the end of the string, where {@literal -1}
+	 * is the last byte, {@literal -2} is the penultimate.
+	 * 
+	 * @param key the key holding the actual String.
+	 * @param bit the bit value to look for.
+	 * @param range must not be {@literal null}. Use {@link Range#unbounded()} to not limit search.
+	 * @return {@literal null} when used in pipeline / transaction. The position of the first bit set to 1 or 0 according
+	 *         to the request.
+	 * @see <a href="http://redis.io/commands/bitpos">Redis Documentation: BITPOS</a>
+	 * @since 2.1
+	 */
+	@Nullable
+	Long bitPos(byte[] key, boolean bit, Range<Long> range);
+
 	/**
 	 * Get the length of the value stored at {@code key}.
 	 *

src/main/java/org/springframework/data/redis/connection/StringRedisConnection.java

@@ -580,6 +580,37 @@ interface StringTuple extends Tuple {
 	 */
 	Long bitOp(BitOperation op, String destination, String... keys);
 
+	/**
+	 * Return the position of the first bit set to given {@code bit} in a string.
+	 *
+	 * @param key the key holding the actual String.
+	 * @param bit the bit value to look for.
+	 * @return {@literal null} when used in pipeline / transaction. The position of the first bit set to 1 or 0 according
+	 *         to the request.
+	 * @see <a href="http://redis.io/commands/bitpos">Redis Documentation: BITPOS</a>
+	 * @since 2.1
+	 */
+	default Long bitPos(String key, boolean bit) {
+		return bitPos(key, bit, org.springframework.data.domain.Range.unbounded());
+	}
+
+	/**
+	 * Return the position of the first bit set to given {@code bit} in a string.
+	 * {@link org.springframework.data.domain.Range} start and end can contain negative values in order to index
+	 * <strong>bytes</strong> starting from the end of the string, where {@literal -1} is the last byte, {@literal -2} is
+	 * the penultimate.
+	 *
+	 * @param key the key holding the actual String.
+	 * @param bit the bit value to look for.
+	 * @param range must not be {@literal null}. Use {@link Range#unbounded()} to not limit search.
+	 * @return {@literal null} when used in pipeline / transaction. The position of the first bit set to 1 or 0 according
+	 *         to the request.
+	 * @see <a href="http://redis.io/commands/bitpos">Redis Documentation: BITPOS</a>
+	 * @since 2.1
+	 */
+	@Nullable
+	Long bitPos(String key, boolean bit, org.springframework.data.domain.Range<Long> range);
+
 	/**
 	 * Get the length of the value stored at {@code key}.
 	 *

src/main/java/org/springframework/data/redis/connection/jedis/JedisClusterStringCommands.java

@@ -19,18 +19,21 @@
 import lombok.RequiredArgsConstructor;
 import redis.clients.jedis.BinaryJedis;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.ClusterSlotHashUtil;
 import org.springframework.data.redis.connection.RedisStringCommands;
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.connection.jedis.JedisClusterConnection.JedisClusterCommandCallback;
 import org.springframework.data.redis.connection.jedis.JedisClusterConnection.JedisMultiKeyClusterCommandCallback;
+import org.springframework.data.redis.connection.lettuce.LettuceConverters;
 import org.springframework.data.redis.core.types.Expiration;
 import org.springframework.data.redis.util.ByteUtils;
 import org.springframework.util.Assert;
@@ -489,6 +492,29 @@ public Long bitOp(BitOperation op, byte[] destination, byte[]... keys) {
 		throw new InvalidDataAccessApiUsageException("BITOP is only supported for same slot keys in cluster mode.");
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitPos(byte[], boolean, org.springframework.data.Range)
+	 */
+	@Override
+	public Long bitPos(byte[] key, boolean bit, Range<Long> range) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(range, "Range must not be null! Use Range.unbounded() instead.");
+
+		List<byte[]> args = new ArrayList<>(3);
+		args.add(LettuceConverters.toBit(bit));
+
+		if (range.getLowerBound().isBounded()) {
+			args.add(range.getLowerBound().getValue().map(LettuceConverters::toBytes).get());
+		}
+		if (range.getUpperBound().isBounded()) {
+			args.add(range.getUpperBound().getValue().map(LettuceConverters::toBytes).get());
+		}
+
+		return Long.class.cast(connection.execute("BITPOS", key, args));
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#strLen(byte[])

src/main/java/org/springframework/data/redis/connection/jedis/JedisStringCommands.java

@@ -17,14 +17,17 @@
 
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
+import redis.clients.jedis.BitPosParams;
 
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.data.domain.Range;
 import org.springframework.data.redis.connection.RedisStringCommands;
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.core.types.Expiration;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
 
@@ -713,6 +716,43 @@ public Long bitOp(BitOperation op, byte[] destination, byte[]... keys) {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitOp(byte[], boolean, org.springframework.data.domain.Range)
+	 */
+	@Nullable
+	@Override
+	public Long bitPos(byte[] key, boolean bit, Range<Long> range) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(range, "Range must not be null! Use Range.unbounded() instead.");
+
+		BitPosParams params = null;
+		if (range.getLowerBound().isBounded()) {
+			params = range.getUpperBound().isBounded()
+					? new BitPosParams(range.getLowerBound().getValue().get(), range.getUpperBound().getValue().get())
+					: new BitPosParams(range.getLowerBound().getValue().get());
+		}
+
+		try {
+			if (isPipelined()) {
+
+				pipeline(connection.newJedisResult(params != null ? connection.getRequiredPipeline().bitpos(key, bit, params)
+						: connection.getRequiredPipeline().bitpos(key, bit)));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(
+						connection.newJedisResult(params != null ? connection.getRequiredTransaction().bitpos(key, bit, params)
+								: connection.getRequiredTransaction().bitpos(key, bit)));
+				return null;
+			}
+			return params != null ? connection.getJedis().bitpos(key, bit, params) : connection.getJedis().bitpos(key, bit);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#strLen(byte[])