DATAREDIS-562 - Polishing.

Move BitfieldCommand to top-level type BitFieldSubCommands.
Use primitives in command objects. Add toString methods. Add utility methods to extract responses from Jedis Client. Use Connection.execute(…) to invoke BITFIELD using Jedis. Fix Overflow values to string rendering.
Convert test ticket references to new format.

Original pull request: #227.

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

@@ -0,0 +1,620 @@
+/*
+ * Copyright 2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.redis.connection;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldSubCommand;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * The actual {@code BITFIELD} command representation holding several {@link BitFieldSubCommand}s to execute.
+ *
+ * @author Christoph Strobl
+ * @since 2.1
+ */
+public class BitFieldSubCommands implements Iterable<BitFieldSubCommand> {
+
+	private final List<BitFieldSubCommand> subCommands;
+
+	private BitFieldSubCommands(List<BitFieldSubCommand> subCommands) {
+		this.subCommands = new ArrayList<>(subCommands);
+	}
+
+	private BitFieldSubCommands(List<BitFieldSubCommand> subCommands, BitFieldSubCommand subCommand) {
+
+		this(subCommands);
+
+		Assert.notNull(subCommand, "SubCommand must not be null!");
+		this.subCommands.add(subCommand);
+	}
+
+	/**
+	 * Creates a new {@link BitFieldSubCommands}.
+	 *
+	 * @return
+	 */
+	public static BitFieldSubCommands create() {
+		return new BitFieldSubCommands(Collections.emptyList());
+	}
+
+	/**
+	 * Obtain a new {@link BitFieldGetBuilder} for creating and adding a {@link BitFieldGet} sub command.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @return
+	 */
+	public BitFieldGetBuilder get(BitFieldType type) {
+		return new BitFieldGetBuilder(this).forType(type);
+	}
+
+	/**
+	 * Create new {@link BitFieldSubCommands} adding given {@link BitFieldGet} to the sub commands.
+	 *
+	 * @param get must not be {@literal null}.
+	 * @return
+	 */
+	protected BitFieldSubCommands get(BitFieldGet get) {
+		return new BitFieldSubCommands(subCommands, get);
+	}
+
+	/**
+	 * Obtain a new {@link BitFieldSetBuilder} for creating and adding a {@link BitFieldSet} sub command.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @return
+	 */
+	public BitFieldSetBuilder set(BitFieldType type) {
+		return new BitFieldSetBuilder(this).forType(type);
+	}
+
+	/**
+	 * Create new {@link BitFieldSubCommands} adding given {@link BitFieldSet} to the sub commands.
+	 *
+	 * @param get must not be {@literal null}.
+	 * @return
+	 */
+	protected BitFieldSubCommands set(BitFieldSet set) {
+		return new BitFieldSubCommands(subCommands, set);
+	}
+
+	/**
+	 * Obtain a new {@link BitFieldIncrByBuilder} for creating and adding a {@link BitFieldIncrby} sub command.
+	 *
+	 * @param type must not be {@literal null}.
+	 * @return
+	 */
+	public BitFieldIncrByBuilder incr(BitFieldType type) {
+		return new BitFieldIncrByBuilder(this).forType(type);
+	}
+
+	/**
+	 * Create new {@link BitFieldSubCommands} adding given {@link BitFieldIncrBy} to the sub commands.
+	 *
+	 * @param get must not be {@literal null}.
+	 * @return
+	 */
+	protected BitFieldSubCommands incr(BitFieldIncrBy incrBy) {
+		return new BitFieldSubCommands(subCommands, incrBy);
+	}
+
+	/**
+	 * Get the {@link List} of sub commands.
+	 *
+	 * @return never {@literal null}.
+	 */
+	public List<BitFieldSubCommand> getSubCommands() {
+		return subCommands;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see java.lang.Iterable#iterator()
+	 */
+	@Override
+	public Iterator<BitFieldSubCommand> iterator() {
+		return subCommands.iterator();
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 */
+	public static class BitFieldSetBuilder {
+
+		private BitFieldSubCommands ref;
+
+		BitFieldSet set = new BitFieldSet();
+
+		private BitFieldSetBuilder(BitFieldSubCommands ref) {
+			this.ref = ref;
+		}
+
+		public BitFieldSetBuilder forType(BitFieldType type) {
+			this.set.type = type;
+			return this;
+		}
+
+		/**
+		 * Set the zero based bit {@literal offset}.
+		 *
+		 * @param offset must not be {@literal null}.
+		 * @return
+		 */
+		public BitFieldSetBuilder valueAt(long offset) {
+			return valueAt(Offset.offset(offset));
+		}
+
+		/**
+		 * Set the bit offset.
+		 *
+		 * @param offset must not be {@literal null}.
+		 * @return
+		 */
+		public BitFieldSetBuilder valueAt(Offset offset) {
+
+			Assert.notNull(offset, "Offset must not be null!");
+
+			this.set.offset = offset;
+			return this;
+		}
+
+		/**
+		 * Set the value.
+		 *
+		 * @param value must not be {@literal null}.
+		 * @return
+		 */
+		public BitFieldSubCommands to(long value) {
+
+			this.set.value = value;
+			return ref.set(this.set);
+		}
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 */
+	public class BitFieldGetBuilder {
+
+		private BitFieldSubCommands ref;
+
+		BitFieldGet get = new BitFieldGet();
+
+		private BitFieldGetBuilder(BitFieldSubCommands ref) {
+			this.ref = ref;
+		}
+
+		public BitFieldGetBuilder forType(BitFieldType type) {
+			this.get.type = type;
+			return this;
+		}
+
+		/**
+		 * Set the zero based bit {@literal offset}.
+		 *
+		 * @param offset must not be {@literal null}.
+		 * @return
+		 */
+		public BitFieldSubCommands valueAt(long offset) {
+			return valueAt(Offset.offset(offset));
+		}
+
+		/**
+		 * Set the bit offset.
+		 *
+		 * @param offset must not be {@literal null}.
+		 * @return
+		 */
+		public BitFieldSubCommands valueAt(Offset offset) {
+
+			Assert.notNull(offset, "Offset must not be null!");
+
+			this.get.offset = offset;
+			return ref.get(this.get);
+		}
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 */
+	public class BitFieldIncrByBuilder {
+
+		private BitFieldSubCommands ref;
+
+		BitFieldIncrBy incrBy = new BitFieldIncrBy();
+
+		private BitFieldIncrByBuilder(BitFieldSubCommands ref) {
+			this.ref = ref;
+		}
+
+		public BitFieldIncrByBuilder forType(BitFieldType type) {
+			this.incrBy.type = type;
+			return this;
+		}
+
+		/**
+		 * Set the zero based bit {@literal offset}.
+		 *
+		 * @param offset must not be {@literal null}.
+		 * @return
+		 */
+		public BitFieldIncrByBuilder valueAt(long offset) {
+			return valueAt(Offset.offset(offset));
+		}
+
+		/**
+		 * Set the bit offset.
+		 *
+		 * @param offset must not be {@literal null}.
+		 * @return
+		 */
+		public BitFieldIncrByBuilder valueAt(Offset offset) {
+
+			Assert.notNull(offset, "Offset must not be null!");
+			this.incrBy.offset = offset;
+			return this;
+		}
+
+		/**
+		 * Set the {@link BitFieldIncrBy.Overflow} to be used for this and any subsequent {@link BitFieldIncrBy} commands.
+		 *
+		 * @param overflow
+		 * @return
+		 */
+		public BitFieldIncrByBuilder overflow(BitFieldIncrBy.Overflow overflow) {
+			this.incrBy.overflow = overflow;
+			return this;
+		}
+
+		/**
+		 * Set the value used for increasing.
+		 *
+		 * @param value
+		 * @return
+		 */
+		public BitFieldSubCommands by(long value) {
+
+			this.incrBy.value = value;
+			return ref.incr(this.incrBy);
+		}
+	}
+
+	/**
+	 * Sub command to be used as part of {@link BitFieldSubCommands}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	public interface BitFieldSubCommand {
+
+		/**
+		 * The actual sub command
+		 *
+		 * @return never {@literal null}.
+		 */
+		String getCommand();
+
+		/**
+		 * The {@link BitFieldType} to apply for the command.
+		 *
+		 * @return never {@literal null}.
+		 */
+		BitFieldType getType();
+
+		/**
+		 * The bit offset to apply for the command.
+		 *
+		 * @return never {@literal null}.
+		 */
+		Offset getOffset();
+	}
+
+	/**
+	 * Offset used inside a {@link BitFieldSubCommand}. Can be zero or type based. See
+	 * <a href="http://redis.io/commands/bitfield#bits-and-positional-offsets">Bits and positional offsets</a> in the
+	 * Redis reference.
+	 *
+	 * @author Christoph Strobl
+	 * @author Mark Paluch
+	 * @since 2.1
+	 */
+	public static class Offset {
+
+		private final long offset;
+		private final boolean zeroBased;
+
+		private Offset(long offset, boolean zeroBased) {
+
+			this.offset = offset;
+			this.zeroBased = zeroBased;
+		}
+
+		/**
+		 * Creates new zero based offset. <br />
+		 * <b>NOTE:</b> change to type based offset by calling {@link #multipliedByTypeLength()}.
+		 *
+		 * @param offset must not be {@literal null}.
+		 * @return
+		 */
+		public static Offset offset(long offset) {
+			return new Offset(offset, true);
+		}
+
+		/**
+		 * Creates new type based offset.
+		 *
+		 * @return
+		 */
+		public Offset multipliedByTypeLength() {
+			return new Offset(offset, false);
+		}
+
+		/**
+		 * @return true if offset starts at 0 and is not multiplied by the type length.
+		 */
+		public boolean isZeroBased() {
+			return zeroBased;
+		}
+
+		/**
+		 * @return the actual offset value
+		 */
+		public long getValue() {
+			return offset;
+		}
+
+		/**
+		 * @return the Redis Command representation
+		 */
+		public String asString() {
+			return (isZeroBased() ? "" : "#") + getValue();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see java.lang.Object#toString()
+		 */
+		@Override
+		public String toString() {
+			return asString();
+		}
+	}
+
+	/**
+	 * The actual Redis bitfield type representation for signed and unsigned integers used with
+	 * {@link BitFieldSubCommand}.
+	 *
+	 * @author Christoph Strobl
+	 * @author Mark Paluch
+	 * @since 2.1
+	 */
+	public static class BitFieldType {
+
+		/** 8 bit signed Integer */
+		public static final BitFieldType INT_8 = new BitFieldType(true, 8);
+
+		/** 16 bit signed Integer */
+		public static final BitFieldType INT_16 = new BitFieldType(true, 16);
+
+		/** 32 bit signed Integer */
+		public static final BitFieldType INT_32 = new BitFieldType(true, 32);
+
+		/** 64 bit signed Integer */
+		public static final BitFieldType INT_64 = new BitFieldType(true, 64);
+
+		/** 8 bit unsigned Integer */
+		public static final BitFieldType UINT_8 = new BitFieldType(false, 8);
+
+		/** 16 bit unsigned Integer */
+		public static final BitFieldType UINT_16 = new BitFieldType(false, 16);
+
+		/** 32 bit unsigned Integer */
+		public static final BitFieldType UINT_32 = new BitFieldType(false, 32);
+
+		/** 64 bit unsigned Integer */
+		public static final BitFieldType UINT_64 = new BitFieldType(false, 64);
+
+		private final boolean signed;
+		private final int bits;
+
+		private BitFieldType(Boolean signed, Integer bits) {
+
+			this.signed = signed;
+			this.bits = bits;
+		}
+
+		/**
+		 * Create new signed {@link BitFieldType}.
+		 *
+		 * @param bits must not be {@literal null}.
+		 * @return
+		 */
+		public static BitFieldType signed(int bits) {
+			return new BitFieldType(false, bits);
+		}
+
+		/**
+		 * Create new unsigned {@link BitFieldType}.
+		 *
+		 * @param bits must not be {@literal null}.
+		 * @return
+		 */
+		public static BitFieldType unsigned(int bits) {
+			return new BitFieldType(false, bits);
+		}
+
+		/**
+		 * @return true if {@link BitFieldType} is signed.
+		 */
+		public boolean isSigned() {
+			return signed;
+		}
+
+		/**
+		 * Get the actual bits of the type.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public int getBits() {
+			return bits;
+		}
+
+		/**
+		 * Get the Redis Command representation.
+		 *
+		 * @return
+		 */
+		public String asString() {
+			return (isSigned() ? "i" : "u") + getBits();
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see java.lang.Object#toString()
+		 */
+		@Override
+		public String toString() {
+			return asString();
+		}
+	}
+
+	/**
+	 * @author Christoph Strobl
+	 */
+	public static abstract class AbstractBitFieldSubCommand implements BitFieldSubCommand {
+
+		BitFieldType type;
+		Offset offset;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitFieldSubCommand#getType()
+		 */
+		@Override
+		public BitFieldType getType() {
+			return type;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitFieldSubCommand#getOffset()
+		 */
+		@Override
+		public Offset getOffset() {
+			return offset;
+		}
+	}
+
+	/**
+	 * The {@code SET} sub command used with {@link BitFieldSubCommands}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	public static class BitFieldSet extends AbstractBitFieldSubCommand {
+
+		private long value;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitFieldSubCommand#getCommand()
+		 */
+		@Override
+		public String getCommand() {
+			return "SET";
+		}
+
+		/**
+		 * Get the value to set.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public long getValue() {
+			return value;
+		}
+
+	}
+
+	/**
+	 * The {@code GET} sub command used with {@link BitFieldSubCommands}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	public static class BitFieldGet extends AbstractBitFieldSubCommand {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitFieldSubCommand#getCommand()
+		 */
+		@Override
+		public String getCommand() {
+			return "GET";
+		}
+
+	}
+
+	/**
+	 * The {@code INCRBY} sub command used with {@link BitFieldSubCommands}.
+	 *
+	 * @author Christoph Strobl
+	 * @since 2.1
+	 */
+	public static class BitFieldIncrBy extends AbstractBitFieldSubCommand {
+
+		private long value;
+		private @Nullable Overflow overflow;
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitFieldSubCommand#getCommand()
+		 */
+		@Override
+		public String getCommand() {
+			return "INCRBY";
+		}
+
+		/**
+		 * Get the increment value.
+		 *
+		 * @return never {@literal null}.
+		 */
+		public long getValue() {
+			return value;
+		}
+
+		/**
+		 * Get the overflow to apply. Can be {@literal null} to use redis defaults.
+		 *
+		 * @return can be {@literal null}.
+		 */
+		@Nullable
+		public Overflow getOverflow() {
+			return overflow;
+		}
+
+		/**
+		 * @author Christoph Strobl
+		 */
+		public enum Overflow {
+			SAT, FAIL, WRAP
+		}
+	}
+}

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

@@ -27,6 +27,7 @@ import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Queue;
 import java.util.Set;
+import java.util.Map.Entry;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.commons.logging.Log;
@@ -3662,8 +3663,8 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
 	 */
 	@Override
-	public List<Long> bitfield(byte[] key, BitfieldCommand operation) {
-		return delegate.bitfield(key, operation);
+	public List<Long> bitField(byte[] key, BitFieldSubCommands subCommands) {
+		return delegate.bitField(key, subCommands);
 	}
 
 	/*
@@ -3671,9 +3672,9 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 	 * @see org.springframework.data.redis.connection.StringRedisConnection#bitfield(byte[], BitfieldCommand)
 	 */
 	@Override
-	public List<Long> bitfield(String key, BitfieldCommand operation) {
+	public List<Long> bitfield(String key, BitFieldSubCommands operation) {
 
-		List<Long> results = delegate.bitfield(serialize(key), operation);
+		List<Long> results = delegate.bitField(serialize(key), operation);
 		if (isFutureConversion()) {
 			addResultConverter(identityConverter);
 		}

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

@@ -23,7 +23,6 @@ import java.util.Properties;
 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;
@@ -404,8 +403,8 @@ public interface DefaultedRedisConnection extends RedisConnection {
 	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
 	@Override
 	@Deprecated
-	default List<Long> bitfield(byte[] key, BitfieldCommand operation) {
-		return stringCommands().bitfield(key, operation);
+	default List<Long> bitField(byte[] key, BitFieldSubCommands subCommands) {
+		return stringCommands().bitField(key, subCommands);
 	}
 
 	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */

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

@@ -15,15 +15,11 @@
  */
 package org.springframework.data.redis.connection;
 
-import java.util.ArrayList;
-import java.util.Collections;
 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.util.Assert;
-import org.springframework.util.ObjectUtils;
 import org.springframework.lang.Nullable;
 
 /**
@@ -285,6 +281,18 @@ public interface RedisStringCommands {
 	@Nullable
 	Long bitCount(byte[] key, long start, long end);
 
+	/**
+	 * 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 must not be {@literal null}.
+	 * @return {@literal null} when used in pipeline / transaction.
+	 * @since 2.1
+	 */
+	@Nullable
+	List<Long> bitField(byte[] key, BitFieldSubCommands subCommands);
+
 	/**
 	 * Perform bitwise operations between strings.
 	 *
@@ -338,18 +346,6 @@ public interface RedisStringCommands {
 	@Nullable
 	Long strLen(byte[] key);
 
-	/**
-	 * 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 operation must not be {@literal null}.
-	 * @return {@literal null} when used in pipeline / transaction.
-	 * @since 2.1
-	 */
-	@Nullable
-	List<Long> bitfield(byte[] key, BitfieldCommand operation);
-
 	/**
 	 * {@code SET} command arguments for {@code NX}, {@code XX}.
 	 *
@@ -406,581 +402,4 @@ public interface RedisStringCommands {
 			return SET_IF_ABSENT;
 		}
 	}
-
-	/**
-	 * The actual {@code BITFIELD} command representation holding several {@link BitfieldSubCommand}s to execute.
-	 *
-	 * @author Christoph Strobl
-	 * @since 1.8
-	 */
-	class BitfieldCommand {
-
-		private final List<BitfieldSubCommand> subCommands;
-
-		private BitfieldCommand(List<BitfieldSubCommand> subCommands) {
-
-			this.subCommands = new ArrayList<BitfieldSubCommand>(
-					subCommands != null ? subCommands : Collections.<BitfieldSubCommand> emptyList());
-		}
-
-		private BitfieldCommand(List<BitfieldSubCommand> subCommands, BitfieldSubCommand subCommand) {
-
-			this(subCommands);
-
-			Assert.notNull(subCommand, "SubCommand must not be null!");
-			this.subCommands.add(subCommand);
-		}
-
-		/**
-		 * Creates a new {@link BitfieldCommand}.
-		 *
-		 * @return
-		 */
-		public static BitfieldCommand newBitfieldCommand() {
-			return new BitfieldCommand(Collections.<BitfieldSubCommand> emptyList());
-		}
-
-		/**
-		 * Obtain a new {@link BitfieldGetBuilder} for creating and adding a {@link BitfieldGet} sub command.
-		 *
-		 * @param type must not be {@literal null}.
-		 * @return
-		 */
-		public BitfieldGetBuilder get(BitfieldType type) {
-			return new BitfieldGetBuilder(this).forType(type);
-		}
-
-		/**
-		 * Create new {@link BitfieldCommand} adding given {@link BitfieldGet} to the sub commands.
-		 *
-		 * @param get must not be {@literal null}.
-		 * @return
-		 */
-		protected BitfieldCommand get(BitfieldGet get) {
-			return new BitfieldCommand(subCommands, get);
-		}
-
-		/**
-		 * Obtain a new {@link BitfieldSetBuilder} for creating and adding a {@link BitfieldSet} sub command.
-		 *
-		 * @param type must not be {@literal null}.
-		 * @return
-		 */
-		public BitfieldSetBuilder set(BitfieldType type) {
-			return new BitfieldSetBuilder(this).forType(type);
-		}
-
-		/**
-		 * Create new {@link BitfieldCommand} adding given {@link BitfieldSet} to the sub commands.
-		 *
-		 * @param get must not be {@literal null}.
-		 * @return
-		 */
-		protected BitfieldCommand set(BitfieldSet set) {
-			return new BitfieldCommand(subCommands, set);
-		}
-
-		/**
-		 * Obtain a new {@link BitfieldIncrByBuilder} for creating and adding a {@link BitfieldIncrby} sub command.
-		 *
-		 * @param type must not be {@literal null}.
-		 * @return
-		 */
-		public BitfieldIncrByBuilder incr(BitfieldType type) {
-			return new BitfieldIncrByBuilder(this).forType(type);
-		}
-
-		/**
-		 * Create new {@link BitfieldCommand} adding given {@link BitfieldIncrBy} to the sub commands.
-		 *
-		 * @param get must not be {@literal null}.
-		 * @return
-		 */
-		protected BitfieldCommand incr(BitfieldIncrBy incrBy) {
-			return new BitfieldCommand(subCommands, incrBy);
-		}
-
-		/**
-		 * Get the {@link List} of sub commands.
-		 *
-		 * @return never {@literal null}.
-		 */
-		public List<BitfieldSubCommand> getSubCommands() {
-			return subCommands;
-		}
-
-		/**
-		 * @author Christoph Strobl
-		 */
-		public class BitfieldSetBuilder {
-
-			private BitfieldCommand ref;
-
-			BitfieldSet set = new BitfieldSet();
-
-			private BitfieldSetBuilder(BitfieldCommand ref) {
-				this.ref = ref;
-			}
-
-			public BitfieldSetBuilder forType(BitfieldType type) {
-				this.set.type = type;
-				return this;
-			}
-
-			/**
-			 * Set the zero based bit {@literal offset}.
-			 *
-			 * @param offset must not be {@literal null}.
-			 * @return
-			 */
-			public BitfieldSetBuilder valueAt(Long offset) {
-				return valueAt(Offset.offset(offset));
-			}
-
-			/**
-			 * Set the bit offset.
-			 *
-			 * @param offset must not be {@literal null}.
-			 * @return
-			 */
-			public BitfieldSetBuilder valueAt(Offset offset) {
-
-				Assert.notNull(offset, "Offset must not be null!");
-
-				this.set.offset = offset;
-				return this;
-			}
-
-			/**
-			 * Set the value.
-			 *
-			 * @param value must not be {@literal null}.
-			 * @return
-			 */
-			public BitfieldCommand to(Long value) {
-
-				Assert.notNull(value, "Value must not be null!");
-
-				this.set.value = value;
-				return ref.set(this.set);
-			}
-		}
-
-		/**
-		 * @author Christoph Strobl
-		 */
-		public class BitfieldGetBuilder {
-
-			private BitfieldCommand ref;
-
-			BitfieldGet get = new BitfieldGet();
-
-			private BitfieldGetBuilder(BitfieldCommand ref) {
-				this.ref = ref;
-			}
-
-			public BitfieldGetBuilder forType(BitfieldType type) {
-				this.get.type = type;
-				return this;
-			}
-
-			/**
-			 * Set the zero based bit {@literal offset}.
-			 *
-			 * @param offset must not be {@literal null}.
-			 * @return
-			 */
-			public BitfieldCommand valueAt(Long offset) {
-				return valueAt(Offset.offset(offset));
-			}
-
-			/**
-			 * Set the bit offset.
-			 *
-			 * @param offset must not be {@literal null}.
-			 * @return
-			 */
-			public BitfieldCommand valueAt(Offset offset) {
-
-				Assert.notNull(offset, "Offset must not be null!");
-
-				this.get.offset = offset;
-				return ref.get(this.get);
-			}
-		}
-
-		/**
-		 * @author Christoph Strobl
-		 */
-		public class BitfieldIncrByBuilder {
-
-			private BitfieldCommand ref;
-
-			BitfieldIncrBy incrBy = new BitfieldIncrBy();
-
-			private BitfieldIncrByBuilder(BitfieldCommand ref) {
-				this.ref = ref;
-			}
-
-			public BitfieldIncrByBuilder forType(BitfieldType type) {
-				this.incrBy.type = type;
-				return this;
-			}
-
-			/**
-			 * Set the zero based bit {@literal offset}.
-			 *
-			 * @param offset must not be {@literal null}.
-			 * @return
-			 */
-			public BitfieldIncrByBuilder valueAt(Long offset) {
-				return valueAt(Offset.offset(offset));
-			}
-
-			/**
-			 * Set the bit offset.
-			 *
-			 * @param offset must not be {@literal null}.
-			 * @return
-			 */
-			public BitfieldIncrByBuilder valueAt(Offset offset) {
-
-				Assert.notNull(offset, "Offset must not be null!");
-				this.incrBy.offset = offset;
-				return this;
-			}
-
-			/**
-			 * Set the {@link BitfieldIncrBy.Overflow} to be used for this and any subsequent {@link BitfieldIncrBy} commands.
-			 *
-			 * @param overflow
-			 * @return
-			 */
-			public BitfieldIncrByBuilder overflow(BitfieldIncrBy.Overflow overflow) {
-				this.incrBy.overflow = overflow;
-				return this;
-			}
-
-			/**
-			 * Set the value used for increasing.
-			 *
-			 * @param value
-			 * @return
-			 */
-			public BitfieldCommand by(Long value) {
-
-				Assert.notNull(value, "Value must not be null!");
-
-				this.incrBy.value = value;
-				return ref.incr(this.incrBy);
-			}
-		}
-	}
-
-	/**
-	 * Sub command to be used as part of {@link BitfieldCommand}.
-	 *
-	 * @author Christoph Strobl
-	 * @since 1.8
-	 */
-	interface BitfieldSubCommand {
-
-		/**
-		 * The actual sub command
-		 *
-		 * @return never {@literal null}.
-		 */
-		String getCommand();
-
-		/**
-		 * The {@link BitfieldType} to apply for the command.
-		 *
-		 * @return never {@literal null}.
-		 */
-		BitfieldType getType();
-
-		/**
-		 * The bit offset to apply for the command.
-		 *
-		 * @return never {@literal null}.
-		 */
-		Offset getOffset();
-	}
-
-	/**
-	 * Offset used inside a {@link BitfieldSubCommand}. Can be zero or type based. See
-	 * <a href="http://redis.io/commands/bitfield#bits-and-positional-offsets">Bits and positional offsets</a> in the
-	 * Redis reference.
-	 *
-	 * @author Christoph Strobl
-	 * @since 1.8
-	 */
-	class Offset {
-
-		private final Long offset;
-		private final Boolean zeroBased;
-
-		private Offset(Long offset, Boolean zeroBased) {
-
-			Assert.notNull(offset, "Offset must not be null!");
-			Assert.notNull(zeroBased, "ZeroBased must not be null!");
-
-			this.offset = offset;
-			this.zeroBased = zeroBased;
-		}
-
-		/**
-		 * Creates new zero based offset. <br />
-		 * <b>NOTE:</b> change to type based offset by calling {@link #multipliedByTypeLength()}.
-		 *
-		 * @param offset must not be {@literal null}.
-		 * @return
-		 */
-		public static Offset offset(Long offset) {
-			return new Offset(offset, Boolean.TRUE);
-		}
-
-		/**
-		 * Creates new type based offset.
-		 *
-		 * @return
-		 */
-		public Offset multipliedByTypeLength() {
-			return new Offset(offset, Boolean.FALSE);
-		}
-
-		/**
-		 * @return true if offset starts at 0 and is not multiplied by the type length.
-		 */
-		public Boolean isZeroBased() {
-			return zeroBased;
-		}
-
-		/**
-		 * @return the actual offset value
-		 */
-		public Long getValue() {
-			return offset;
-		}
-
-		/**
-		 * @return the Redis Command representation
-		 */
-		public String asString() {
-			return (isZeroBased() ? "" : "#") + getValue();
-		}
-	}
-
-	/**
-	 * The actual Redis bitfield type representation for signed and unsigned integers used with
-	 * {@link BitfieldSubCommand}.
-	 *
-	 * @author Christoph Strobl
-	 * @since 1.8
-	 */
-	class BitfieldType {
-
-		/** 8 bit signed Integer */
-		public static final BitfieldType INT_8 = new BitfieldType(Boolean.TRUE, 8);
-
-		/** 16 bit signed Integer */
-		public static final BitfieldType INT_16 = new BitfieldType(Boolean.TRUE, 16);
-
-		/** 32 bit signed Integer */
-		public static final BitfieldType INT_32 = new BitfieldType(Boolean.TRUE, 32);
-
-		/** 64 bit signed Integer */
-		public static final BitfieldType INT_64 = new BitfieldType(Boolean.TRUE, 64);
-
-		/** 8 bit unsigned Integer */
-		public static final BitfieldType UINT_8 = new BitfieldType(Boolean.FALSE, 8);
-
-		/** 16 bit unsigned Integer */
-		public static final BitfieldType UINT_16 = new BitfieldType(Boolean.FALSE, 16);
-
-		/** 32 bit unsigned Integer */
-		public static final BitfieldType UINT_32 = new BitfieldType(Boolean.FALSE, 32);
-
-		/** 64 bit unsigned Integer */
-		public static final BitfieldType UINT_64 = new BitfieldType(Boolean.FALSE, 64);
-
-		private final Boolean signed;
-		private final Integer bits;
-
-		private BitfieldType(Boolean signed, Integer bits) {
-
-			Assert.notNull(signed, "Signed must not be null!");
-			Assert.notNull(bits, "Bits must not be null!");
-
-			this.signed = signed;
-			this.bits = bits;
-		}
-
-		/**
-		 * Create new signed {@link BitfieldType}.
-		 *
-		 * @param bits must not be {@literal null}.
-		 * @return
-		 */
-		public static BitfieldType signed(Integer bits) {
-			return new BitfieldType(Boolean.TRUE, bits);
-		}
-
-		/**
-		 * Create new unsigned {@link BitfieldType}.
-		 *
-		 * @param bits must not be {@literal null}.
-		 * @return
-		 */
-		public static BitfieldType unsigned(Integer bits) {
-			return new BitfieldType(Boolean.FALSE, bits);
-		}
-
-		/**
-		 * @return true if {@link BitfieldType} is signed.
-		 */
-		public boolean isSigned() {
-			return ObjectUtils.nullSafeEquals(signed, Boolean.TRUE);
-		}
-
-		/**
-		 * Get the actual bits of the type.
-		 *
-		 * @return never {@literal null}.
-		 */
-		public Integer getBits() {
-			return bits;
-		}
-
-		/**
-		 * Get the Redis Command representation.
-		 *
-		 * @return
-		 */
-		public String asString() {
-			return (isSigned() ? "i" : "u") + getBits();
-		}
-	}
-
-	/**
-	 * @author Christoph Strobl
-	 */
-	abstract class AbstractBitfieldSubCommand implements BitfieldSubCommand {
-
-		BitfieldType type;
-		Offset offset;
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitfieldSubCommand#getType()
-		 */
-		@Override
-		public BitfieldType getType() {
-			return type;
-		}
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitfieldSubCommand#getOffset()
-		 */
-		@Override
-		public Offset getOffset() {
-			return offset;
-		}
-	}
-
-	/**
-	 * The {@code SET} sub command used with {@link BitfieldCommand}.
-	 *
-	 * @author Christoph Strobl
-	 * @since 1.8
-	 */
-	class BitfieldSet extends AbstractBitfieldSubCommand {
-
-		private Long value;
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitfieldSubCommand#getCommand()
-		 */
-		@Override
-		public String getCommand() {
-			return "SET";
-		}
-
-		/**
-		 * Get the value to set.
-		 *
-		 * @return never {@literal null}.
-		 */
-		public Long getValue() {
-			return value;
-		}
-
-	}
-
-	/**
-	 * The {@code GET} sub command used with {@link BitfieldCommand}.
-	 *
-	 * @author Christoph Strobl
-	 * @since 1.8
-	 */
-	class BitfieldGet extends AbstractBitfieldSubCommand {
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitfieldSubCommand#getCommand()
-		 */
-		@Override
-		public String getCommand() {
-			return "GET";
-		}
-
-	}
-
-	/**
-	 * The {@code INCRBY} sub command used with {@link BitfieldCommand}.
-	 *
-	 * @author Christoph Strobl
-	 * @since 1.8
-	 */
-	class BitfieldIncrBy extends AbstractBitfieldSubCommand {
-
-		private Long value;
-		private Overflow overflow;
-
-		/*
-		 * (non-Javadoc)
-		 * @see org.springframework.data.redis.connection.StringRedisConnection.BitfieldSubCommand#getCommand()
-		 */
-		@Override
-		public String getCommand() {
-			return "INCRBY";
-		}
-
-		/**
-		 * Get the increment value.
-		 *
-		 * @return never {@literal null}.
-		 */
-		public Long getValue() {
-			return value;
-		}
-
-		/**
-		 * Get the overflow to apply. Can be {@literal null} to use redis defaults.
-		 *
-		 * @return can be {@literal null}.
-		 */
-		public Overflow getOverflow() {
-			return overflow;
-		}
-
-		/**
-		 * @author Christoph Strobl
-		 */
-		public enum Overflow {
-			SAT, FAIL, WRAP
-		}
-	}
 }

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

@@ -1954,5 +1954,5 @@ public interface StringRedisConnection extends RedisConnection {
 	 * @param command must not be {@literal null}.
 	 * @return
 	 */
-	List<Long> bitfield(String key, BitfieldCommand command);
+	List<Long> bitfield(String key, BitFieldSubCommands command);
 }

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

@@ -31,6 +31,7 @@ import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.util.Arrays;
 import java.util.Set;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -104,12 +105,28 @@ class JedisClientUtils {
 	 * @return the response, can be be {@literal null}.
 	 */
 	static <T> T execute(String command, byte[][] keys, byte[][] args, Supplier<Jedis> jedis) {
+		return execute(command, keys, args, jedis, it -> (T) it.getOne());
+	}
+
+	/**
+	 * Execute an arbitrary on the supplied {@link Jedis} instance.
+	 *
+	 * @param command the command.
+	 * @param keys must not be {@literal null}, may be empty.
+	 * @param args must not be {@literal null}, may be empty.
+	 * @param jedis must not be {@literal null}.
+	 * @param responseMapper must not be {@literal null}.
+	 * @return the response, can be be {@literal null}.
+	 * @since 2.1
+	 */
+	static <T> T execute(String command, byte[][] keys, byte[][] args, Supplier<Jedis> jedis,
+			Function<Client, T> responseMapper) {
 
 		byte[][] commandArgs = getCommandArguments(keys, args);
 
 		Client client = sendCommand(command, commandArgs, jedis.get());
 
-		return (T) client.getOne();
+		return responseMapper.apply(client);
 	}
 
 	/**

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

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.jedis;
 
 import redis.clients.jedis.BinaryJedis;
 import redis.clients.jedis.BinaryJedisPubSub;
+import redis.clients.jedis.Client;
 import redis.clients.jedis.HostAndPort;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisCluster;
@@ -30,6 +31,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.function.Function;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -151,6 +153,11 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 	@Nullable
 	@Override
 	public <T> T execute(String command, byte[] key, Collection<byte[]> args) {
+		return execute(command, key, args, it -> (T) it.getOne());
+	}
+
+	@Nullable
+	<T> T execute(String command, byte[] key, Collection<byte[]> args, Function<Client, T> responseMapper) {
 
 		Assert.notNull(command, "Command must not be null!");
 		Assert.notNull(key, "Key must not be null!");
@@ -161,7 +168,7 @@ public class JedisClusterConnection implements DefaultedRedisClusterConnection {
 		RedisClusterNode keyMaster = topologyProvider.getTopology().getKeyServingMasterNode(key);
 
 		return clusterCommandExecutor.executeCommandOnSingleNode((JedisClusterCommandCallback<T>) client -> JedisClientUtils
-				.execute(command, EMPTY_2D_BYTE_ARRAY, commandArgs, () -> client), keyMaster).getValue();
+				.execute(command, EMPTY_2D_BYTE_ARRAY, commandArgs, () -> client, responseMapper), keyMaster).getValue();
 	}
 
 	private static byte[][] getCommandArguments(byte[] key, Collection<byte[]> args) {

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

@@ -18,6 +18,7 @@ package org.springframework.data.redis.connection.jedis;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import redis.clients.jedis.BinaryJedis;
+import redis.clients.jedis.Connection;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -28,6 +29,7 @@ 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.BitFieldSubCommands;
 import org.springframework.data.redis.connection.ClusterSlotHashUtil;
 import org.springframework.data.redis.connection.RedisStringCommands;
 import org.springframework.data.redis.connection.convert.Converters;
@@ -474,16 +476,15 @@ class JedisClusterStringCommands implements RedisStringCommands {
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
 	 */
 	@Override
-	public List<Long> bitfield(byte[] key, BitfieldCommand command) {
+	public List<Long> bitField(byte[] key, BitFieldSubCommands subCommands) {
 
 		Assert.notNull(key, "Key must not be null!");
-		Assert.notNull(command, "Command must not be null!");
+		Assert.notNull(subCommands, "Command must not be null!");
+
+		byte[][] args = JedisConverters.toBitfieldCommandArguments(subCommands);
 
 		try {
-
-			List untypedListToAvoidClassCastErrorsSinceThisOneDeclaresListOfByteArrayButReturnsListOfLong = connection.getCluster()
-					.bitfield(key, JedisConverters.toBitfieldCommandArguments(command));
-			return (List<Long>) untypedListToAvoidClassCastErrorsSinceThisOneDeclaresListOfByteArrayButReturnsListOfLong;
+			return connection.execute("BITFIELD", key, Arrays.asList(args), Connection::getIntegerMultiBulkReply);
 		} catch (Exception ex) {
 			throw convertJedisAccessException(ex);
 		}

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

@@ -17,6 +17,7 @@ package org.springframework.data.redis.connection.jedis;
 
 import redis.clients.jedis.BinaryJedisPubSub;
 import redis.clients.jedis.Client;
+import redis.clients.jedis.Connection;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.Pipeline;
 import redis.clients.jedis.Response;
@@ -29,6 +30,7 @@ import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Queue;
+import java.util.function.Function;
 import java.util.function.Supplier;
 
 import org.springframework.core.convert.converter.Converter;
@@ -241,6 +243,11 @@ public class JedisConnection extends AbstractRedisConnection {
 	 */
 	@Override
 	public Object execute(String command, byte[]... args) {
+		return execute(command, args, Connection::getOne, JedisClientUtils::getResponse);
+	}
+
+	<T> T execute(String command, byte[][] args, Function<Client, T> resultMapper,
+			Function<Object, Response<?>> pipelineResponseMapper) {
 
 		Assert.hasText(command, "A valid command needs to be specified!");
 		Assert.notNull(args, "Arguments must not be null!");
@@ -251,8 +258,8 @@ public class JedisConnection extends AbstractRedisConnection {
 
 			if (isQueueing() || isPipelined()) {
 
-				Response<Object> result = JedisClientUtils
-						.getResponse(isPipelined() ? getRequiredPipeline() : getRequiredTransaction());
+				Response<?> result = pipelineResponseMapper
+						.apply(isPipelined() ? getRequiredPipeline() : getRequiredTransaction());
 				if (isPipelined()) {
 					pipeline(newJedisResult(result));
 				} else {
@@ -260,7 +267,7 @@ public class JedisConnection extends AbstractRedisConnection {
 				}
 				return null;
 			}
-			return client.getOne();
+			return resultMapper.apply(client);
 		} catch (Exception ex) {
 			throw convertJedisAccessException(ex);
 		}

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

@@ -43,6 +43,10 @@ import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.geo.Metrics;
 import org.springframework.data.geo.Point;
+import org.springframework.data.redis.connection.BitFieldSubCommands;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldIncrBy;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldSet;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldSubCommand;
 import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.RedisClusterNode;
 import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
@@ -104,7 +108,7 @@ abstract public class JedisConverters extends Converters {
 	private static final Converter<byte[], String> BYTES_TO_STRING_CONVERTER;
 	private static final ListConverter<byte[], String> BYTES_LIST_TO_STRING_LIST_CONVERTER;
 	private static final ListConverter<byte[], Long> BYTES_LIST_TO_LONG_LIST_CONVERTER;
-	private static final Converter<RedisStringCommands.BitfieldCommand, List<byte[]>> BITFIELD_COMMAND_ARGUMENT_CONVERTER;
+	private static final Converter<BitFieldSubCommands, List<byte[]>> BITFIELD_COMMAND_ARGUMENT_CONVERTER;
 
 	public static final byte[] PLUS_BYTES;
 	public static final byte[] MINUS_BYTES;
@@ -201,9 +205,9 @@ abstract public class JedisConverters extends Converters {
 			}
 		});
 
-		BITFIELD_COMMAND_ARGUMENT_CONVERTER = new Converter<RedisStringCommands.BitfieldCommand, List<byte[]>>() {
+		BITFIELD_COMMAND_ARGUMENT_CONVERTER = new Converter<BitFieldSubCommands, List<byte[]>>() {
 			@Override
-			public List<byte[]> convert(RedisStringCommands.BitfieldCommand source) {
+			public List<byte[]> convert(BitFieldSubCommands source) {
 
 				if (source == null) {
 					return Collections.emptyList();
@@ -211,11 +215,11 @@ abstract public class JedisConverters extends Converters {
 
 				List<byte[]> args = new ArrayList<byte[]>(source.getSubCommands().size() * 4);
 
-				for (RedisStringCommands.BitfieldSubCommand command : source.getSubCommands()) {
+				for (BitFieldSubCommand command : source.getSubCommands()) {
 
-					if (command instanceof RedisStringCommands.BitfieldIncrBy) {
+					if (command instanceof BitFieldIncrBy) {
 
-						RedisStringCommands.BitfieldIncrBy.Overflow overflow = ((RedisStringCommands.BitfieldIncrBy) command)
+						BitFieldIncrBy.Overflow overflow = ((BitFieldIncrBy) command)
 								.getOverflow();
 						if (overflow != null) {
 							args.add(JedisConverters.toBytes("OVERFLOW"));
@@ -227,10 +231,10 @@ abstract public class JedisConverters extends Converters {
 					args.add(JedisConverters.toBytes(command.getType().asString()));
 					args.add(JedisConverters.toBytes(command.getOffset().asString()));
 
-					if (command instanceof RedisStringCommands.BitfieldSet) {
-						args.add(JedisConverters.toBytes(((RedisStringCommands.BitfieldSet) command).getValue()));
-					} else if (command instanceof RedisStringCommands.BitfieldIncrBy) {
-						args.add(JedisConverters.toBytes(((RedisStringCommands.BitfieldIncrBy) command).getValue()));
+					if (command instanceof BitFieldSet) {
+						args.add(JedisConverters.toBytes(((BitFieldSet) command).getValue()));
+					} else if (command instanceof BitFieldIncrBy) {
+						args.add(JedisConverters.toBytes(((BitFieldIncrBy) command).getValue()));
 					}
 				}
 
@@ -689,14 +693,13 @@ abstract public class JedisConverters extends Converters {
 	}
 
 	/**
-	 * Convert given {@link org.springframework.data.redis.connection.RedisStringCommands.BitfieldCommand} into argument
-	 * array.
+	 * Convert given {@link BitFieldSubCommands} into argument array.
 	 *
 	 * @param bitfieldOperation
 	 * @return never {@literal null}.
 	 * @since 1.8
 	 */
-	public static byte[][] toBitfieldCommandArguments(RedisStringCommands.BitfieldCommand bitfieldOperation) {
+	public static byte[][] toBitfieldCommandArguments(BitFieldSubCommands bitfieldOperation) {
 
 		List<byte[]> tmp = BITFIELD_COMMAND_ARGUMENT_CONVERTER.convert(bitfieldOperation);
 		return tmp.toArray(new byte[tmp.size()][]);

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

@@ -18,15 +18,18 @@ package org.springframework.data.redis.connection.jedis;
 import lombok.NonNull;
 import lombok.RequiredArgsConstructor;
 import redis.clients.jedis.BitPosParams;
+import redis.clients.jedis.Client;
 
 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.BitFieldSubCommands;
 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.data.redis.util.ByteUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -690,33 +693,14 @@ class JedisStringCommands implements RedisStringCommands {
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
 	 */
 	@Override
-	public List<Long> bitfield(byte[] key, BitfieldCommand command) {
+	public List<Long> bitField(byte[] key, BitFieldSubCommands subCommands) {
 
 		Assert.notNull(key, "Key must not be null!");
-		Assert.notNull(command, "Command must not be null!");
+		Assert.notNull(subCommands, "Command must not be null!");
 
-		byte[][] args = JedisConverters.toBitfieldCommandArguments(command);
+		byte[][] args = ByteUtils.mergeArrays(key, JedisConverters.toBitfieldCommandArguments(subCommands));
 
-		try {
-			if (isPipelined()) {
-				pipeline(new JedisResult(connection.getRequiredPipeline().bitfield(key, args)));
-				return null;
-			}
-			if (isQueueing()) {
-				transaction(new JedisResult(connection.getRequiredTransaction().bitfield(key, args)));
-				return null;
-			}
-		} catch (Exception ex) {
-			throw convertJedisAccessException(ex);
-		}
-
-		// TODO: fix type declaration once https://github.com/xetorthio/jedis/issues/1413 is resolved
-		// Jedis return type declaration is wrong. it indicates List<byte[]> but return List<Long> causing Class cast errors
-		// when trying to convert that stuff. So we hacked it here to make it work.
-
-		List untypedListToAvoidClassCastErrorsSinceThisOneDeclaresListOfByteArrayButReturnsListOfLong = connection.getJedis().bitfield(key,
-				args);
-		return (List<Long>) untypedListToAvoidClassCastErrorsSinceThisOneDeclaresListOfByteArrayButReturnsListOfLong;
+		return connection.execute("BITFIELD", args, Client::getIntegerMultiBulkReply, JedisClientUtils::getResponse);
 	}
 
 	/*

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

@@ -33,6 +33,12 @@ import org.springframework.data.geo.GeoResults;
 import org.springframework.data.geo.Metric;
 import org.springframework.data.geo.Metrics;
 import org.springframework.data.geo.Point;
+import org.springframework.data.redis.connection.BitFieldSubCommands;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldGet;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldIncrBy;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldIncrBy.Overflow;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldSet;
+import org.springframework.data.redis.connection.BitFieldSubCommands.BitFieldSubCommand;
 import org.springframework.data.redis.connection.DefaultTuple;
 import org.springframework.data.redis.connection.RedisClusterNode;
 import org.springframework.data.redis.connection.RedisClusterNode.Flag;
@@ -812,6 +818,66 @@ abstract public class LettuceConverters extends Converters {
 		return geoArgs;
 	}
 
+	/**
+	 * Convert {@link BitFieldSubCommands} into {@link BitFieldArgs}.
+	 *
+	 * @param subCommands
+	 * @return
+	 * @since 2.1
+	 */
+	public static BitFieldArgs toBitFieldArgs(BitFieldSubCommands subCommands) {
+
+		BitFieldArgs args = new BitFieldArgs();
+
+		for (BitFieldSubCommand subCommand : subCommands) {
+
+			BitFieldArgs.BitFieldType bft = subCommand.getType().isSigned()
+					? BitFieldArgs.signed(subCommand.getType().getBits())
+					: BitFieldArgs.unsigned(subCommand.getType().getBits());
+
+			BitFieldArgs.Offset offset;
+			if (subCommand.getOffset().isZeroBased()) {
+				offset = BitFieldArgs.offset((int) subCommand.getOffset().getValue());
+			} else {
+				offset = BitFieldArgs.typeWidthBasedOffset((int) subCommand.getOffset().getValue());
+			}
+
+			if (subCommand instanceof BitFieldGet) {
+				args = args.get(bft, offset);
+			} else if (subCommand instanceof BitFieldSet) {
+				args = args.set(bft, offset, ((BitFieldSet) subCommand).getValue());
+			} else if (subCommand instanceof BitFieldIncrBy) {
+
+				BitFieldIncrBy.Overflow overflow = ((BitFieldIncrBy) subCommand).getOverflow();
+				if (overflow != null) {
+
+					BitFieldArgs.OverflowType type;
+
+					switch (overflow) {
+						case SAT:
+							type = BitFieldArgs.OverflowType.SAT;
+							break;
+						case FAIL:
+							type = BitFieldArgs.OverflowType.FAIL;
+							break;
+						case WRAP:
+							type = BitFieldArgs.OverflowType.WRAP;
+							break;
+						default:
+							throw new IllegalArgumentException(
+									String.format("Invalid OVERFLOW. Expected one the following %s but got %s.",
+											Arrays.toString(Overflow.values()), overflow));
+					}
+					args = args.overflow(type);
+				}
+
+				args = args.incrBy(bft, (int) subCommand.getOffset().getValue(), ((BitFieldIncrBy) subCommand).getValue());
+			}
+		}
+
+		return args;
+	}
+
 	/**
 	 * Get {@link Converter} capable of {@link Set} of {@link Byte} into {@link GeoResults}.
 	 *

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

@@ -154,7 +154,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#setEX(org.reactivestreams.Publisher, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#setEX(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<BooleanResponse<SetCommand>> setEX(Publisher<SetCommand> commands) {
@@ -171,7 +171,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#pSetEX(org.reactivestreams.Publisher, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#pSetEX(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<BooleanResponse<SetCommand>> pSetEX(Publisher<SetCommand> commands) {
@@ -237,7 +237,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#getRange(org.reactivestreams.Publisher, java.util.function.Supplier, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#getRange(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<ByteBufferResponse<RangeCommand>> getRange(Publisher<RangeCommand> commands) {
@@ -256,7 +256,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#setRange(org.reactivestreams.Publisher, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#setRange(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<NumericResponse<SetRangeCommand, Long>> setRange(Publisher<SetRangeCommand> commands) {
@@ -274,7 +274,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#getBit(org.reactivestreams.Publisher, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#getBit(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<BooleanResponse<GetBitCommand>> getBit(Publisher<GetBitCommand> commands) {
@@ -291,7 +291,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#setBit(org.reactivestreams.Publisher, java.util.function.Supplier, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#setBit(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<BooleanResponse<SetBitCommand>> setBit(Publisher<SetBitCommand> commands) {
@@ -309,7 +309,7 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#bitCount(org.reactivestreams.Publisher, java.util.function.Supplier, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#bitCount(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<NumericResponse<BitCountCommand, Long>> bitCount(Publisher<BitCountCommand> commands) {
@@ -328,7 +328,16 @@ class LettuceReactiveStringCommands implements ReactiveStringCommands {
 
 	/*
 	 * (non-Javadoc)
-	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#bitOp(org.reactivestreams.Publisher, java.util.function.Supplier, java.util.function.Supplier)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#bitCount(org.reactivestreams.Publisher)
+	 */
+	@Override
+	public Flux<NumericResponse<BitFieldCommand, Long>> bitField(Publisher<BitFieldCommand> commands) {
+		return null;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.ReactiveRedisConnection.ReactiveStringCommands#bitOp(org.reactivestreams.Publisher)
 	 */
 	@Override
 	public Flux<NumericResponse<BitOpCommand, Long>> bitOp(Publisher<BitOpCommand> commands) {

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

@@ -16,7 +16,6 @@
 package org.springframework.data.redis.connection.lettuce;
 
 import io.lettuce.core.BitFieldArgs;
-import io.lettuce.core.BitFieldArgs.Offset;
 import io.lettuce.core.RedisFuture;
 import io.lettuce.core.cluster.api.async.RedisClusterAsyncCommands;
 import io.lettuce.core.cluster.api.sync.RedisClusterCommands;
@@ -29,6 +28,7 @@ import java.util.concurrent.Future;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.BitFieldSubCommands;
 import org.springframework.data.redis.connection.RedisStringCommands;
 import org.springframework.data.redis.connection.convert.Converters;
 import org.springframework.data.redis.core.types.Expiration;
@@ -610,57 +610,12 @@ class LettuceStringCommands implements RedisStringCommands {
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
 	 */
 	@Override
-	public List<Long> bitfield(byte[] key, BitfieldCommand command) {
+	public List<Long> bitField(byte[] key, BitFieldSubCommands subCommands) {
 
 		Assert.notNull(key, "Key must not be null!");
-		Assert.notNull(command, "Command must not be null!");
+		Assert.notNull(subCommands, "Command must not be null!");
 
-		BitFieldArgs args = new BitFieldArgs();
-
-		for (BitfieldSubCommand subCommand : command.getSubCommands()) {
-
-			BitFieldArgs.BitFieldType bft = subCommand.getType().isSigned() ? BitFieldArgs.signed(subCommand.getType().getBits())
-					: BitFieldArgs.unsigned(subCommand.getType().getBits());
-
-			BitFieldArgs.Offset offset;
-			if (!subCommand.getOffset().isZeroBased()) {
-				offset = BitFieldArgs.offset(subCommand.getOffset().getValue().intValue());
-			}else{
-				offset = BitFieldArgs.typeWidthBasedOffset(subCommand.getOffset().getValue().intValue());
-			}
-
-			if (subCommand instanceof BitfieldGet) {
-				args = args.get(bft, offset);
-			} else if (subCommand instanceof BitfieldSet) {
-				args = args.set(bft, offset, ((BitfieldSet) subCommand).getValue());
-			} else if (subCommand instanceof BitfieldIncrBy) {
-
-				BitfieldIncrBy.Overflow overflow = ((BitfieldIncrBy) subCommand).getOverflow();
-				if (overflow != null) {
-
-					BitFieldArgs.OverflowType type;
-
-					switch (overflow) {
-						case SAT:
-							type = BitFieldArgs.OverflowType.SAT;
-							break;
-						case FAIL:
-							type = BitFieldArgs.OverflowType.FAIL;
-							break;
-						case WRAP:
-							type = BitFieldArgs.OverflowType.WRAP;
-							break;
-						default:
-							throw new IllegalArgumentException(
-									String.format("o_O invalid OVERFLOW. Expected one the following %s but got %s.",
-											BitfieldIncrBy.Overflow.values(), overflow));
-					}
-					args = args.overflow(type);
-				}
-
-				args = args.incrBy(bft, subCommand.getOffset().getValue().intValue(), ((BitfieldIncrBy) subCommand).getValue());
-			}
-		}
+		BitFieldArgs args = LettuceConverters.toBitFieldArgs(subCommands);
 
 		try {
 			if (isPipelined()) {

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

@@ -23,8 +23,8 @@ import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.connection.BitFieldSubCommands;
 import org.springframework.data.redis.connection.RedisConnection;
-import org.springframework.data.redis.connection.RedisStringCommands;
 import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
 import org.springframework.data.redis.core.types.Expiration;
 import org.springframework.lang.Nullable;
@@ -393,14 +393,9 @@ class DefaultValueOperations<K, V> extends AbstractOperations<K, V> implements V
 	 * @see org.springframework.data.redis.core.ValueOperations#bitfield(Object, RedisStringCommands.BitfieldCommand)
 	 */
 	@Override
-	public List<Long> bitfield(K key, final RedisStringCommands.BitfieldCommand command) {
+	public List<Long> bitField(K key, final BitFieldSubCommands subCommands) {
 
-		final byte[] rawKey = rawKey(key);
-		return execute(new RedisCallback<List<Long>>() {
-
-			public List<Long> doInRedis(RedisConnection connection) {
-				return connection.bitfield(rawKey, command);
-			}
-		}, true);
+		byte[] rawKey = rawKey(key);
+		return execute(connection -> connection.bitField(rawKey, subCommands), true);
 	}
 }

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

@@ -21,8 +21,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
-import org.springframework.data.redis.connection.RedisStringCommands;
-
+import org.springframework.data.redis.connection.BitFieldSubCommands;
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -354,11 +353,13 @@ public interface ValueOperations<K, V> {
 	 * at a given {@code key}.
 	 *
 	 * @param key must not be {@literal null}.
-	 * @param command must not be {@literal null}.
-	 * @return
+	 * @param subCommands must not be {@literal null}.
+	 * @return {@literal null} when used in pipeline / transaction.
 	 * @since 2.1
+	 * @see <a href="http://redis.io/commands/bitfield">Redis Documentation: BITFIELD</a>
 	 */
-	List<Long> bitfield(K key, RedisStringCommands.BitfieldCommand command);
+	@Nullable
+	List<Long> bitField(K key, BitFieldSubCommands subCommands);
 
 	RedisOperations<K, V> getOperations();
 }