DATAREDIS-562 - Add reactive BitField support.

Original pull request: #227.

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

@@ -902,6 +902,86 @@ public interface ReactiveStringCommands {
 	 */
 	Flux<NumericResponse<BitCountCommand, Long>> bitCount(Publisher<BitCountCommand> commands);
 
+	/**
+	 * {@code BITFIELD} command parameters.
+	 *
+	 * @author Mark Paluch
+	 * @see <a href="http://redis.io/commands/bitfield">Redis Documentation: BITFIELD</a>
+	 * @since 2.1
+	 */
+	class BitFieldCommand extends KeyCommand {
+
+		private @Nullable BitFieldSubCommands subcommands;
+
+		private BitFieldCommand(ByteBuffer key, @Nullable BitFieldSubCommands subcommands) {
+
+			super(key);
+
+			this.subcommands = subcommands;
+		}
+
+		/**
+		 * Creates a new {@link BitFieldCommand} given a {@literal key}.
+		 *
+		 * @param key must not be {@literal null}.
+		 * @return a new {@link BitFieldCommand} for a {@literal key}.
+		 */
+		public static BitFieldCommand bitField(ByteBuffer key) {
+
+			Assert.notNull(key, "Key must not be null!");
+
+			return new BitFieldCommand(key, null);
+		}
+
+		/**
+		 * Applies the {@link BitFieldSubCommands}. Constructs a new command instance with all previously configured
+		 * properties.
+		 *
+		 * @param commands must not be {@literal null}.
+		 * @return a new {@link BitFieldSubCommands} with {@link BitFieldSubCommands} applied.
+		 */
+		public BitFieldCommand commands(BitFieldSubCommands commands) {
+
+			Assert.notNull(commands, "BitFieldCommands must not be null!");
+
+			return new BitFieldCommand(getKey(), commands);
+		}
+
+		public BitFieldSubCommands getSubCommands() {
+			return subcommands;
+		}
+	}
+
+	/**
+	 * Get / Manipulate specific integer fields of varying bit widths and arbitrary non (necessary) aligned offset stored
+	 * at a given {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param subCommands
+	 * @return
+	 * @see <a href="http://redis.io/commands/bitfield">Redis Documentation: BITFIELD</a>
+	 * @since 2.1
+	 */
+	default Mono<List<Long>> bitField(ByteBuffer key, BitFieldSubCommands subCommands) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(subCommands, "BitFieldSubCommands must not be null!");
+
+		return bitField(Mono.just(BitFieldCommand.bitField(key).commands(subCommands))).map(CommandResponse::getOutput)
+				.next();
+	}
+
+	/**
+	 * Get / Manipulate specific integer fields of varying bit widths and arbitrary non (necessary) aligned offset stored
+	 * at a given {@code key}.
+	 *
+	 * @param commands must not be {@literal null}.
+	 * @return
+	 * @see <a href="http://redis.io/commands/bitfield">Redis Documentation: BITFIELD</a>
+	 * @since 2.1
+	 */
+	Flux<MultiValueResponse<BitFieldCommand, Long>> bitField(Publisher<BitFieldCommand> commands);
+
 	/**
 	 * {@code BITOP} command parameters.
 	 *

src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStringCommands.java

@@ -15,12 +15,14 @@
  */
 package org.springframework.data.redis.connection.lettuce;
 
+import io.lettuce.core.BitFieldArgs;
 import io.lettuce.core.SetArgs;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.nio.ByteBuffer;
 import java.util.List;
+import java.util.stream.Collectors;
 
 import org.reactivestreams.Publisher;
 import org.springframework.data.domain.Range;
@@ -328,11 +330,20 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#bitCount(org.reactivestreams.Publisher)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#bitField(org.reactivestreams.Publisher)
 	 */
 	@Override
-	public Flux<NumericResponse<BitFieldCommand, Long>> bitField(Publisher<BitFieldCommand> commands) {
-		return null;
+	public Flux<MultiValueResponse<BitFieldCommand, Long>> bitField(Publisher<BitFieldCommand> commands) {
+
+		return connection.execute(cmd -> Flux.from(commands).concatMap(command -> {
+
+			Assert.notNull(command.getKey(), "Key must not be null!");
+
+			BitFieldArgs args = LettuceConverters.toBitFieldArgs(command.getSubCommands());
+
+			return cmd.bitfield(command.getKey(), args).collectList().map(value -> new MultiValueResponse<>(command,
+					value.stream().map(v -> v.getValueOrElse(null)).collect(Collectors.toList())));
+		}));
 	}
 
 	/*

src/main/java/org/springframework/data/redis/core/DefaultReactiveValueOperations.java

@@ -29,6 +29,7 @@ import java.util.Map;
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+import org.springframework.data.redis.connection.BitFieldSubCommands;
 import org.springframework.data.redis.connection.ReactiveStringCommands;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
 import org.springframework.data.redis.core.types.Expiration;
@@ -318,6 +319,18 @@ class DefaultReactiveValueOperations<K, V> implements ReactiveValueOperations<K,
 		return createMono(connection -> connection.getBit(rawKey(key), offset));
 	}
 
+	/* (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ReactiveValueOperations#bitField(java.lang.Object, org.springframework.data.redis.connection.BitFieldSubCommands)
+	 */
+	@Override
+	public Mono<List<Long>> bitField(K key, BitFieldSubCommands subCommands) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(subCommands, "BitFieldSubCommands must not be null!");
+
+		return createMono(connection -> connection.bitField(rawKey(key), subCommands));
+	}
+
 	/* (non-Javadoc)
 	 * @see org.springframework.data.redis.core.ReactiveValueOperations#delete(java.lang.Object)
 	 */

src/main/java/org/springframework/data/redis/core/ReactiveValueOperations.java

@@ -22,6 +22,8 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 
+import org.springframework.data.redis.connection.BitFieldSubCommands;
+
 /**
  * Reactive Redis operations for simple (or in Redis terminology 'string') values.
  *
@@ -235,6 +237,17 @@ public interface ReactiveValueOperations<K, V> {
 	 */
 	Mono<Boolean> getBit(K key, long offset);
 
+	/**
+	 * Get / Manipulate specific integer fields of varying bit widths and arbitrary non (necessary) aligned offset stored
+	 * at a given {@code key}.
+	 *
+	 * @param key must not be {@literal null}.
+	 * @param command must not be {@literal null}.
+	 * @return
+	 * @since 2.1
+	 */
+	Mono<List<Long>> bitField(K key, BitFieldSubCommands command);
+
 	/**
 	 * Removes the given {@literal key}.
 	 *