Add Profunctor and Contravariant instance for Consumer

monix-reactive/shared/src/main/scala/monix/reactive/Consumer.scala

@@ -17,9 +17,11 @@
 
 package monix.reactive
 
+import cats.Contravariant
+import cats.arrow.Profunctor
 import monix.eval.{Task, TaskLike}
 import monix.execution.cancelables.AssignableCancelable
-import monix.execution.{Cancelable, Callback, Scheduler}
+import monix.execution.{Callback, Cancelable, Scheduler}
 import monix.reactive.internal.consumers._
 import monix.reactive.observers.Subscriber
 
@@ -29,7 +31,7 @@ import monix.reactive.observers.Subscriber
   * being effectively a way to transform observables into
   * [[monix.eval.Task tasks]] for less error prone consuming of streams.
   */
-abstract class Consumer[-In, +R] extends ((Observable[In]) => Task[R])
+abstract class Consumer[-In, +R] extends (Observable[In] => Task[R])
   with Serializable { self =>
 
   /** Builds a new [[monix.reactive.observers.Subscriber Subscriber]]
@@ -401,4 +403,16 @@ object Consumer {
   trait Sync[-In, +R] extends Consumer[In, R] {
     override def createSubscriber(cb: Callback[Throwable, R], s: Scheduler): (Subscriber.Sync[In], AssignableCancelable)
   }
-}
+
+  /** `cats.Contravariant` instance for [[Consumer]]. */
+  implicit def catsContravariant[C]: Contravariant[Consumer[?, C]] = new Contravariant[Consumer[?, C]] {
+    override def contramap[A, B](fa: Consumer[A, C])(f: B => A): Consumer[B, C] =
+      fa.contramap(f)
+  }
+
+  /** `cats.arrow.Profunctor` instance for [[Consumer]]. */
+  implicit val catsProfunctor: Profunctor[Consumer] = new Profunctor[Consumer] {
+    override def dimap[A, B, C, D](fab: Consumer[A, B])(f: C => A)(g: B => D): Consumer[C, D] =
+      fab.contramap(f).map(g)
+  }
+}

monix-reactive/shared/src/test/scala/monix/reactive/BaseTestSuite.scala

@@ -18,8 +18,10 @@
 package monix.reactive
 
 import cats.Eq
+import cats.Monoid
 import minitest.{SimpleTestSuite, TestSuite}
 import minitest.laws.Checkers
+import monix.eval.Task
 import monix.execution.internal.Platform
 import monix.execution.schedulers.TestScheduler
 import monix.reactive.Notification.{OnComplete, OnError, OnNext}
@@ -97,14 +99,6 @@ trait ArbitraryInstances extends ArbitraryInstancesBase with monix.eval.Arbitrar
       }
     }
 
-  implicit def arbitrarySubject[A](implicit arb: Arbitrary[A]): Arbitrary[Subject[A, A]] = Arbitrary {
-    Gen.oneOf(
-      Gen.const(AsyncSubject[A]()),
-      Gen.const(PublishSubject[A]()),
-      arb.arbitrary.map(BehaviorSubject(_)),
-      implicitly[Arbitrary[List[A]]].arbitrary.map(ReplaySubject.create(_)))
-  }
-
   implicit def equalitySubject[A: Arbitrary](implicit A: Eq[A], ec: TestScheduler): Eq[Subject[A, A]] =
     new Eq[Subject[A, A]] {
       def eqv(lh: Subject[A, A], rh: Subject[A, A]): Boolean = {
@@ -122,6 +116,21 @@ trait ArbitraryInstances extends ArbitraryInstancesBase with monix.eval.Arbitrar
         lh.size == rh.size && equalityFuture(eqList, ec).eqv(fa, fb)
       }
     }
+
+  implicit def equalityConsumer[A: Arbitrary](implicit A: Eq[A], ec: TestScheduler): Eq[Consumer[A, A]] =
+    new Eq[Consumer[A, A]] {
+      override def eqv(lh: Consumer[A, A], rh: Consumer[A, A]): Boolean = {
+        val eqList = implicitly[Eq[List[A]]]
+        val arbObservable = implicitly[Arbitrary[Observable[A]]]
+
+        val observable = arbObservable.arbitrary.sample
+
+        val fa = Task.sequence(observable.map(_.consumeWith(lh)).toList).runToFuture
+        val fb = Task.sequence(observable.map(_.consumeWith(rh)).toList).runToFuture
+
+        equalityFuture(eqList, ec).eqv(fa, fb)
+      }
+    }
 }
 
 trait ArbitraryInstancesBase extends monix.eval.ArbitraryInstancesBase {
@@ -139,6 +148,22 @@ trait ArbitraryInstancesBase extends monix.eval.ArbitraryInstancesBase {
     }
   }
 
+  implicit def arbitrarySubject[A](implicit arb: Arbitrary[A]): Arbitrary[Subject[A, A]] = Arbitrary {
+    Gen.oneOf(
+      Gen.const(AsyncSubject[A]()),
+      Gen.const(PublishSubject[A]()),
+      arb.arbitrary.map(BehaviorSubject(_)),
+      implicitly[Arbitrary[List[A]]].arbitrary.map(ReplaySubject.create(_)))
+  }
+
+  implicit def arbitraryConsumer[A](implicit arb: Arbitrary[A], M: Monoid[A]): Arbitrary[Consumer[A, A]] =
+    Arbitrary {
+      Gen.oneOf(
+        Gen.const(Consumer.foldLeft(M.empty)(M.combine)),
+        Gen.const(Consumer.head[A])
+      )
+    }
+
   implicit def cogenForObservable[A]: Cogen[Observable[A]] =
     Cogen[Unit].contramap(_ => ())
 }

monix-reactive/shared/src/test/scala/monix/reactive/TypeClassLawsForConsumerSuite.scala

@@ -0,0 +1,31 @@
+/*
+ * Copyright (c) 2014-2018 by The Monix Project Developers.
+ * See the project homepage at: https://monix.io
+ *
+ * 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 monix.reactive
+
+import cats.laws.discipline.{ContravariantTests, ProfunctorTests}
+
+object TypeClassLawsForConsumerSuite extends BaseLawsTestSuite {
+
+  checkAllAsync("Contravariant[Consumer]") { implicit ec =>
+    ContravariantTests[Consumer[?, Int]].contravariant[Int, Int, Int]
+  }
+
+  checkAllAsync("Profunctor[Consumer]") { implicit ec =>
+    ProfunctorTests[Consumer].profunctor[Int, Int, Int, Int, Int, Int]
+  }
+}