Add producer convenience methods

modules/core/src/main/scala/fs2/kafka/KafkaProducer.scala

@@ -15,6 +15,8 @@ import fs2.kafka.internal._
 import fs2.kafka.internal.converters.collection._
 import org.apache.kafka.clients.producer.RecordMetadata
 import org.apache.kafka.common.{Metric, MetricName}
+import fs2.Chunk
+import cats.Functor
 
 /**
   * [[KafkaProducer]] represents a producer of Kafka records, with the
@@ -56,6 +58,54 @@ abstract class KafkaProducer[F[_], K, V] {
 
 object KafkaProducer {
 
+  implicit class ProducerOps[F[_], K, V](private val producer: KafkaProducer[F, K, V])
+      extends AnyVal {
+
+    /**
+      * Produce a single [[ProducerRecord]] without a passthrough value,
+      * see [[KafkaProducer.produce]] for general semantics.
+      */
+    def produceOne_(record: ProducerRecord[K, V])(implicit F: Functor[F]): F[F[RecordMetadata]] =
+      produceOne(record, ()).map(_.map { res =>
+        res.records.head.get._2 //Should always be present so get is ok
+      })
+
+    /**
+      * Produce a single record to the specified topic using the provided key and value
+      * without a passthrough value, see [[KafkaProducer.produce]] for general semantics.
+      */
+    def produceOne_(topic: String, key: K, value: V)(implicit F: Functor[F]): F[F[RecordMetadata]] =
+      produceOne_(ProducerRecord(topic, key, value))
+
+    /**
+      * Produces the specified [[ProducerRecords]] without a passthrough value,
+      * see [[KafkaProducer.produce]] for general semantics.
+      */
+    def produce_(
+      records: ProducerRecords[K, V, _]
+    )(implicit F: Functor[F]): F[F[Chunk[(ProducerRecord[K, V], RecordMetadata)]]] =
+      producer.produce(records).map(_.map(_.records))
+
+    /**
+      * Produce a single record to the specified topic using the provided key and value,
+      * see [[KafkaProducer.produce]] for general semantics.
+      */
+    def produceOne[P](
+      topic: String,
+      key: K,
+      value: V,
+      passthrough: P
+    ): F[F[ProducerResult[K, V, P]]] =
+      produceOne(ProducerRecord(topic, key, value), passthrough)
+
+    /**
+      * Produce a single [[ProducerRecord]], see [[KafkaProducer.produce]] for general semantics.
+      */
+    def produceOne[P](record: ProducerRecord[K, V], passthrough: P): F[F[ProducerResult[K, V, P]]] =
+      producer.produce(ProducerRecords.one(record, passthrough))
+
+  }
+
   /**
     * [[KafkaProducer.Metrics]] extends [[KafkaProducer]] to provide
     * access to the underlying producer metrics.

modules/core/src/test/scala/fs2/kafka/KafkaProducerSpec.scala

@@ -135,6 +135,118 @@ final class KafkaProducerSpec extends BaseKafkaSpec {
     }
   }
 
+  it("should produce one without passthrough") {
+    withTopic { topic =>
+      createCustomTopic(topic, partitions = 3)
+      val toProduce = ("some-key" -> "some-value")
+
+      val produced =
+        (for {
+          producer <- KafkaProducer.stream(producerSettings[IO])
+          _ <- Stream.eval(IO(producer.toString should startWith("KafkaProducer$")))
+          batched <- Stream
+            .eval(producer.produceOne_(ProducerRecord(topic, toProduce._1, toProduce._2)))
+          _ <- Stream.eval(batched)
+        } yield ()).compile.toVector.unsafeRunSync()
+
+      val consumed =
+        consumeNumberKeyedMessagesFrom[String, String](topic, produced.size)
+
+      consumed should contain only toProduce
+    }
+  }
+
+  it("should produce one from individual topic, key and value without passthrough") {
+    withTopic { topic =>
+      createCustomTopic(topic, partitions = 3)
+      val toProduce = ("some-key" -> "some-value")
+
+      val produced =
+        (for {
+          producer <- KafkaProducer.stream(producerSettings[IO])
+          _ <- Stream.eval(IO(producer.toString should startWith("KafkaProducer$")))
+          batched <- Stream
+            .eval(producer.produceOne_(topic, toProduce._1, toProduce._2))
+          _ <- Stream.eval(batched)
+        } yield ()).compile.toVector.unsafeRunSync()
+
+      val consumed =
+        consumeNumberKeyedMessagesFrom[String, String](topic, produced.size)
+
+      consumed should contain only toProduce
+    }
+  }
+
+  it("should produce one") {
+    withTopic { topic =>
+      createCustomTopic(topic, partitions = 3)
+      val toProduce = ("some-key" -> "some-value")
+      val passthrough = "passthrough"
+
+      val result =
+        (for {
+          producer <- KafkaProducer.stream(producerSettings[IO])
+          _ <- Stream.eval(IO(producer.toString should startWith("KafkaProducer$")))
+          batched <- Stream
+            .eval(
+              producer.produceOne(ProducerRecord(topic, toProduce._1, toProduce._2), passthrough)
+            )
+          result <- Stream.eval(batched)
+        } yield result).compile.lastOrError.unsafeRunSync()
+
+      assert(result.passthrough == passthrough)
+    }
+  }
+
+  it("should produce one from individual topic, key and value") {
+    withTopic { topic =>
+      createCustomTopic(topic, partitions = 3)
+      val toProduce = ("some-key" -> "some-value")
+      val passthrough = "passthrough"
+
+      val result =
+        (for {
+          producer <- KafkaProducer.stream(producerSettings[IO])
+          _ <- Stream.eval(IO(producer.toString should startWith("KafkaProducer$")))
+          batched <- Stream
+            .eval(producer.produceOne(topic, toProduce._1, toProduce._2, passthrough))
+          result <- Stream.eval(batched)
+        } yield result).compile.lastOrError.unsafeRunSync()
+
+      assert(result.passthrough == passthrough)
+    }
+  }
+
+  it("should be able to produce records with multiple without passthrough") {
+    withTopic { topic =>
+      createCustomTopic(topic, partitions = 3)
+      val toProduce = (0 until 10).map(n => s"key-$n" -> s"value->$n").toList
+
+      val produced =
+        (for {
+          producer <- KafkaProducer.stream(producerSettings[IO])
+          records = ProducerRecords(toProduce.map {
+            case (key, value) =>
+              ProducerRecord(topic, key, value)
+          }, ())
+          result <- Stream.eval(producer.produce_(records).flatten)
+        } yield result).compile.lastOrError.unsafeRunSync()
+
+      val records =
+        produced.map {
+          case (record, _) =>
+            record.key -> record.value
+        }.toList
+
+      assert(records == toProduce)
+
+      val consumed =
+        consumeNumberKeyedMessagesFrom[String, String](topic, toProduce.size)
+
+      consumed should contain theSameElementsAs toProduce
+    }
+  }
+
   it("should get metrics") {
     withTopic { topic =>
       createCustomTopic(topic, partitions = 3)