DATAREDIS-562 - Add support for BITFIELD.

We now offer support for BITFIELD via RedisConnection and RedisClusterConnection using Lettuce and Jedis.

connection.bitField(key,
  create()
   .set(INT_8).valueAt(offset(0).multipliedByTypeLength()).to(100)
   .incr(signed(8)).valueAt(offset(102)).overflow(FAIL).by(1));

Original pull request: #227.

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

@@ -3657,6 +3657,29 @@ public class DefaultStringRedisConnection implements StringRedisConnection, Deco
 		return convertedResults;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
+	 */
+	@Override
+	public List<Long> bitfield(byte[] key, BitfieldCommand operation) {
+		return delegate.bitfield(key, operation);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.StringRedisConnection#bitfield(byte[], BitfieldCommand)
+	 */
+	@Override
+	public List<Long> bitfield(String key, BitfieldCommand operation) {
+
+		List<Long> results = delegate.bitfield(serialize(key), operation);
+		if (isFutureConversion()) {
+			addResultConverter(identityConverter);
+		}
+		return results;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.DecoratedRedisConnection#getDelegate()

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

@@ -401,6 +401,13 @@ public interface DefaultedRedisConnection extends RedisConnection {
 		return stringCommands().bitCount(key, start, end);
 	}
 
+	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
+	@Override
+	@Deprecated
+	default List<Long> bitfield(byte[] key, BitfieldCommand operation) {
+		return stringCommands().bitfield(key, operation);
+	}
+
 	/** @deprecated in favor of {@link RedisConnection#stringCommands()}}. */
 	@Override
 	@Deprecated

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

@@ -15,11 +15,15 @@
  */
 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;
 
 /**
@@ -312,7 +316,7 @@ public interface RedisStringCommands {
 	 * 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.
@@ -334,6 +338,18 @@ 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}.
 	 *
@@ -390,4 +406,581 @@ 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

@@ -1945,4 +1945,14 @@ public interface StringRedisConnection extends RedisConnection {
 	 */
 	List<RedisClientInfo> getClientList();
 
+
+	/**
+	 * 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
+	 */
+	List<Long> bitfield(String key, BitfieldCommand command);
 }

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

@@ -469,6 +469,27 @@ class JedisClusterStringCommands implements RedisStringCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
+	 */
+	@Override
+	public List<Long> bitfield(byte[] key, BitfieldCommand command) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(command, "Command must not be null!");
+
+		try {
+
+			List untypedListToAvoidClassCastErrorsSinceThisOneDeclaresListOfByteArrayButReturnsListOfLong = connection.getCluster()
+					.bitfield(key, JedisConverters.toBitfieldCommandArguments(command));
+			return (List<Long>) untypedListToAvoidClassCastErrorsSinceThisOneDeclaresListOfByteArrayButReturnsListOfLong;
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitOp(org.springframework.data.redis.connection.RedisStringCommands.BitOperation, byte[], byte[][])

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

@@ -103,6 +103,8 @@ abstract public class JedisConverters extends Converters {
 	private static final ListConverter<redis.clients.jedis.GeoCoordinate, Point> LIST_GEO_COORDINATE_TO_POINT_CONVERTER;
 	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;
 
 	public static final byte[] PLUS_BYTES;
 	public static final byte[] MINUS_BYTES;
@@ -191,6 +193,50 @@ abstract public class JedisConverters extends Converters {
 		GEO_COORDINATE_TO_POINT_CONVERTER = geoCoordinate -> geoCoordinate != null
 				? new Point(geoCoordinate.getLongitude(), geoCoordinate.getLatitude()) : null;
 		LIST_GEO_COORDINATE_TO_POINT_CONVERTER = new ListConverter<>(GEO_COORDINATE_TO_POINT_CONVERTER);
+
+		BYTES_LIST_TO_LONG_LIST_CONVERTER = new ListConverter<byte[], Long>(new Converter<byte[], Long>() {
+			@Override
+			public Long convert(byte[] source) {
+				return Long.valueOf(JedisConverters.toString(source));
+			}
+		});
+
+		BITFIELD_COMMAND_ARGUMENT_CONVERTER = new Converter<RedisStringCommands.BitfieldCommand, List<byte[]>>() {
+			@Override
+			public List<byte[]> convert(RedisStringCommands.BitfieldCommand source) {
+
+				if (source == null) {
+					return Collections.emptyList();
+				}
+
+				List<byte[]> args = new ArrayList<byte[]>(source.getSubCommands().size() * 4);
+
+				for (RedisStringCommands.BitfieldSubCommand command : source.getSubCommands()) {
+
+					if (command instanceof RedisStringCommands.BitfieldIncrBy) {
+
+						RedisStringCommands.BitfieldIncrBy.Overflow overflow = ((RedisStringCommands.BitfieldIncrBy) command)
+								.getOverflow();
+						if (overflow != null) {
+							args.add(JedisConverters.toBytes("OVERFLOW"));
+							args.add(JedisConverters.toBytes(overflow.name()));
+						}
+					}
+
+					args.add(JedisConverters.toBytes(command.getCommand()));
+					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()));
+					}
+				}
+
+				return args;
+			}
+		};
 	}
 
 	public static Converter<String, byte[]> stringToBytes() {
@@ -548,6 +594,10 @@ abstract public class JedisConverters extends Converters {
 		return BYTES_LIST_TO_STRING_LIST_CONVERTER;
 	}
 
+	public static ListConverter<byte[], Long> getBytesListToLongListConverter() {
+		return BYTES_LIST_TO_LONG_LIST_CONVERTER;
+	}
+
 	/**
 	 * @return
 	 * @since 1.8
@@ -638,6 +688,20 @@ abstract public class JedisConverters extends Converters {
 		return param;
 	}
 
+	/**
+	 * Convert given {@link org.springframework.data.redis.connection.RedisStringCommands.BitfieldCommand} into argument
+	 * array.
+	 *
+	 * @param bitfieldOperation
+	 * @return never {@literal null}.
+	 * @since 1.8
+	 */
+	public static byte[][] toBitfieldCommandArguments(RedisStringCommands.BitfieldCommand bitfieldOperation) {
+
+		List<byte[]> tmp = BITFIELD_COMMAND_ARGUMENT_CONVERTER.convert(bitfieldOperation);
+		return tmp.toArray(new byte[tmp.size()][]);
+	}
+
 	/**
 	 * @author Christoph Strobl
 	 * @since 1.8

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

@@ -685,6 +685,40 @@ class JedisStringCommands implements RedisStringCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
+	 */
+	@Override
+	public List<Long> bitfield(byte[] key, BitfieldCommand command) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(command, "Command must not be null!");
+
+		byte[][] args = JedisConverters.toBitfieldCommandArguments(command);
+
+		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;
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitOp(org.springframework.data.redis.connection.RedisStringCommands.BitOperation, byte[], byte[][])

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

@@ -15,6 +15,8 @@
  */
 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;
@@ -603,6 +605,78 @@ class LettuceStringCommands implements RedisStringCommands {
 		}
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitfield(byte[], BitfieldCommand)
+	 */
+	@Override
+	public List<Long> bitfield(byte[] key, BitfieldCommand command) {
+
+		Assert.notNull(key, "Key must not be null!");
+		Assert.notNull(command, "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());
+			}
+		}
+
+		try {
+			if (isPipelined()) {
+				pipeline(connection.newLettuceResult(getAsyncConnection().bitfield(key, args)));
+				return null;
+			}
+			if (isQueueing()) {
+				transaction(connection.newLettuceResult(getAsyncConnection().bitfield(key, args)));
+				return null;
+			}
+			return getConnection().bitfield(key, args);
+		} catch (Exception ex) {
+			throw convertLettuceAccessException(ex);
+		}
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.redis.connection.RedisStringCommands#bitOp(org.springframework.data.redis.connection.RedisStringCommands.BitOperation, byte[], byte[][])

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

@@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit;
 
 import org.springframework.dao.DataAccessException;
 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;
@@ -386,4 +387,20 @@ class DefaultValueOperations<K, V> extends AbstractOperations<K, V> implements V
 		byte[] rawKey = rawKey(key);
 		return execute(connection -> connection.getBit(rawKey, offset), true);
 	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.redis.core.ValueOperations#bitfield(Object, RedisStringCommands.BitfieldCommand)
+	 */
+	@Override
+	public List<Long> bitfield(K key, final RedisStringCommands.BitfieldCommand command) {
+
+		final byte[] rawKey = rawKey(key);
+		return execute(new RedisCallback<List<Long>>() {
+
+			public List<Long> doInRedis(RedisConnection connection) {
+				return connection.bitfield(rawKey, command);
+			}
+		}, true);
+	}
 }

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

@@ -21,6 +21,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.data.redis.connection.RedisStringCommands;
+
 import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 
@@ -347,6 +349,16 @@ public interface ValueOperations<K, V> {
 	@Nullable
 	Boolean getBit(K key, long offset);
 
-	RedisOperations<K, V> getOperations();
+	/**
+	 * 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
+	 */
+	List<Long> bitfield(K key, RedisStringCommands.BitfieldCommand command);
 
+	RedisOperations<K, V> getOperations();
 }