diff --git a/driver/src/main/scala/api/collections/bson/bsoncollection.scala b/driver/src/main/scala/api/collections/bson/bsoncollection.scala
index 26cbf7ef8..78a6983a1 100644
--- a/driver/src/main/scala/api/collections/bson/bsoncollection.scala
+++ b/driver/src/main/scala/api/collections/bson/bsoncollection.scala
@@ -46,6 +46,10 @@ object BSONBatchCommands extends BatchCommands[BSONSerializationPack.type] {
   implicit def CountWriter = BSONCountCommandImplicits.CountWriter
   implicit def CountResultReader = BSONCountCommandImplicits.CountResultReader
 
+  val DistinctCommand = BSONDistinctCommand
+  implicit def DistinctWriter = BSONDistinctCommandImplicits.DistinctWriter
+  implicit def DistinctResultReader = BSONDistinctCommandImplicits.DistinctResultReader
+
   val InsertCommand = BSONInsertCommand
   implicit def InsertWriter = BSONInsertCommandImplicits.InsertWriter
 
diff --git a/driver/src/main/scala/api/collections/genericcollection.scala b/driver/src/main/scala/api/collections/genericcollection.scala
index c54533057..23878e2b1 100644
--- a/driver/src/main/scala/api/collections/genericcollection.scala
+++ b/driver/src/main/scala/api/collections/genericcollection.scala
@@ -51,7 +51,7 @@ trait GenericCollectionWithCommands[P <: SerializationPack with Singleton] { sel
 }
 
 trait BatchCommands[P <: SerializationPack] {
-  import reactivemongo.api.commands.{ AggregationFramework => AC, CountCommand => CC, InsertCommand => IC, UpdateCommand => UC, DeleteCommand => DC, DefaultWriteResult, LastError, ResolvedCollectionCommand, FindAndModifyCommand => FMC }
+  import reactivemongo.api.commands.{ AggregationFramework => AC, CountCommand => CC, DistinctCommand => DistC, InsertCommand => IC, UpdateCommand => UC, DeleteCommand => DC, DefaultWriteResult, LastError, ResolvedCollectionCommand, FindAndModifyCommand => FMC }
 
   val pack: P
 
@@ -59,6 +59,10 @@ trait BatchCommands[P <: SerializationPack] {
   implicit def CountWriter: pack.Writer[ResolvedCollectionCommand[CountCommand.Count]]
   implicit def CountResultReader: pack.Reader[CountCommand.CountResult]
 
+  val DistinctCommand: DistC[pack.type]
+  implicit def DistinctWriter: pack.Writer[ResolvedCollectionCommand[DistinctCommand.Distinct]]
+  implicit def DistinctResultReader: pack.Reader[DistinctCommand.DistinctResult]
+
   val InsertCommand: IC[pack.type]
   implicit def InsertWriter: pack.Writer[ResolvedCollectionCommand[InsertCommand.Insert]]
 
@@ -175,6 +179,15 @@ trait GenericCollection[P <: SerializationPack with Singleton] extends Collectio
    */
   def count[H](selector: Option[pack.Document] = None, limit: Int = 0, skip: Int = 0, hint: Option[H] = None)(implicit h: H => CountCommand.Hint, ec: ExecutionContext): Future[Int] = runValueCommand(CountCommand.Count(query = selector, limit, skip, hint.map(h)))
 
+  /**
+   * Returns the distinct values for a specified field across a single collection and returns the results in an array.
+   * @param key the field for which to return distinct values
+   * @param selector the query selector that specifies the documents from which to retrieve the distinct values.
+   */
+  def distinct(key: String, selector: Option[pack.Document] = None)(implicit ec: ExecutionContext): Future[List[pack.Value]] = {
+    runCommand(DistinctCommand.Distinct(keyString = key, query = selector)).map(_.values)
+  }
+
   @inline private def defaultWriteConcern = db.connection.options.writeConcern
 
   def bulkInsert(ordered: Boolean)(documents: ImplicitlyDocumentProducer*)(implicit ec: ExecutionContext): Future[MultiBulkWriteResult] =
diff --git a/driver/src/main/scala/api/commands/bson/distinct.scala b/driver/src/main/scala/api/commands/bson/distinct.scala
new file mode 100644
index 000000000..809757311
--- /dev/null
+++ b/driver/src/main/scala/api/commands/bson/distinct.scala
@@ -0,0 +1,27 @@
+package reactivemongo.api.commands.bson
+
+import reactivemongo.api.BSONSerializationPack
+import reactivemongo.api.commands._
+import reactivemongo.bson._
+
+object BSONDistinctCommand extends DistinctCommand[BSONSerializationPack.type] {
+  val pack = BSONSerializationPack
+}
+
+object BSONDistinctCommandImplicits {
+  import BSONDistinctCommand._
+
+  implicit object DistinctWriter extends BSONDocumentWriter[ResolvedCollectionCommand[Distinct]] {
+    def write(distinct: ResolvedCollectionCommand[Distinct]): BSONDocument =
+      BSONDocument(
+        "distinct" -> distinct.collection,
+        "key" -> distinct.command.keyString,
+        "query" -> distinct.command.query)
+  }
+
+  implicit object DistinctResultReader extends DealingWithGenericCommandErrorsReader[DistinctResult] {
+    def readResult(doc: BSONDocument): DistinctResult =
+      DistinctResult(doc.getAs[BSONArray]("values").fold[List[BSONValue]](List())(_.values.toList))
+
+  }
+}
diff --git a/driver/src/main/scala/api/commands/distinct.scala b/driver/src/main/scala/api/commands/distinct.scala
new file mode 100644
index 000000000..c7a46ae71
--- /dev/null
+++ b/driver/src/main/scala/api/commands/distinct.scala
@@ -0,0 +1,12 @@
+package reactivemongo.api.commands
+
+import reactivemongo.api.SerializationPack
+
+trait DistinctCommand[P <: SerializationPack] extends ImplicitCommandHelpers[P] {
+  case class Distinct(
+    //{ distinct: <collection>, key: <key>, query: <query> }
+    keyString: String,
+    query: Option[pack.Document]) extends CollectionCommand with CommandWithPack[pack.type] with CommandWithResult[DistinctResult]
+
+  case class DistinctResult(values: List[pack.Value])
+}
diff --git a/driver/src/test/scala/AggregationSpec.scala b/driver/src/test/scala/AggregationSpec.scala
index ca14b5e03..229d1eabb 100644
--- a/driver/src/test/scala/AggregationSpec.scala
+++ b/driver/src/test/scala/AggregationSpec.scala
@@ -107,5 +107,10 @@ object AggregationSpec extends org.specs2.mutable.Specification {
         map(_.documents) aka "results" must beEqualTo(expected).
         await(timeoutMillis)
     }
+
+    "return distinct states" in {
+      val expected: List[BSONValue] = List("NY", "FR", "JP").map(BSONString.apply)
+      collection.distinct("state").aka("results") must beEqualTo(expected).await(timeoutMillis)
+    }
   }
 }