Allow multiple channels to be created for a single connection

core/src/main/scala/com/github/gvolpe/fs2rabbit/algebra/Channeller.scala

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2017-2019 Gabriel Volpe
+ *
+ * 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 com.github.gvolpe.fs2rabbit.algebra
+
+import com.github.gvolpe.fs2rabbit.model.AMQPChannel
+
+trait Channeller[F[_]] {
+  def createChannel: F[AMQPChannel]
+}

core/src/main/scala/com/github/gvolpe/fs2rabbit/algebra/Connection.scala

@@ -16,8 +16,10 @@
 
 package com.github.gvolpe.fs2rabbit.algebra
 
-import com.github.gvolpe.fs2rabbit.model.AMQPChannel
+import com.github.gvolpe.fs2rabbit.model.{AMQPChannel, AMQPConnection}
 
 trait Connection[F[_]] {
+  def createConnection: F[AMQPConnection]
+
   def createConnectionChannel: F[AMQPChannel]
 }

core/src/main/scala/com/github/gvolpe/fs2rabbit/interpreter/ChannelStream.scala

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2017-2019 Gabriel Volpe
+ *
+ * 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 com.github.gvolpe.fs2rabbit.interpreter
+
+import cats.effect.Sync
+import cats.syntax.apply._
+import cats.syntax.functor._
+import com.github.gvolpe.fs2rabbit.algebra.Channeller
+import com.github.gvolpe.fs2rabbit.effects.Log
+import com.github.gvolpe.fs2rabbit.model
+import com.github.gvolpe.fs2rabbit.model.{AMQPChannel, AMQPConnection, RabbitChannel}
+import fs2.Stream
+
+class ChannelStream[F[_]](
+    connection: AMQPConnection
+)(implicit F: Sync[F], L: Log[F])
+    extends Channeller[Stream[F, ?]] {
+
+  private val acquireChannel: F[AMQPChannel] =
+    for {
+      chan <- F.delay(connection.value.createChannel())
+    } yield RabbitChannel(chan)
+
+  override def createChannel: Stream[F, model.AMQPChannel] =
+    Stream
+      .bracket(acquireChannel) {
+        case RabbitChannel(chan) =>
+          L.info(s"Releasing channel: $chan previously acquired.") *>
+            F.delay { if (chan.isOpen) chan.close() }
+      }
+}

core/src/main/scala/com/github/gvolpe/fs2rabbit/interpreter/ConnectionStream.scala

@@ -16,18 +16,19 @@
 
 package com.github.gvolpe.fs2rabbit.interpreter
 
-import javax.net.ssl.SSLContext
 import cats.data.NonEmptyList
 import cats.effect.Sync
 import cats.syntax.apply._
 import cats.syntax.flatMap._
 import cats.syntax.functor._
 import com.github.gvolpe.fs2rabbit.algebra.Connection
 import com.github.gvolpe.fs2rabbit.config.Fs2RabbitConfig
-import com.github.gvolpe.fs2rabbit.model.{AMQPChannel, RabbitChannel}
 import com.github.gvolpe.fs2rabbit.effects.Log
+import com.github.gvolpe.fs2rabbit.model.{AMQPChannel, AMQPConnection, RabbitChannel, RabbitConnection}
 import com.rabbitmq.client.{Address, ConnectionFactory, Connection => RabbitMQConnection}
 import fs2.Stream
+import javax.net.ssl.SSLContext
+
 import scala.collection.JavaConverters._
 
 class ConnectionStream[F[_]](
@@ -36,19 +37,32 @@ class ConnectionStream[F[_]](
 )(implicit F: Sync[F], L: Log[F])
     extends Connection[Stream[F, ?]] {
 
-  private[fs2rabbit] val acquireConnection: F[(RabbitMQConnection, AMQPChannel)] =
+  private[fs2rabbit] val acquireConnectionChannel: F[(RabbitMQConnection, AMQPChannel)] =
     for {
       conn    <- F.delay(factory.newConnection(addresses.toList.asJava))
       channel <- F.delay(conn.createChannel)
     } yield (conn, RabbitChannel(channel))
 
+  private[fs2rabbit] val acquireConnection: F[AMQPConnection] =
+    for {
+      conn <- F.delay(factory.newConnection(addresses.toList.asJava))
+    } yield RabbitConnection(conn)
+
+  override def createConnection: Stream[F, AMQPConnection] =
+    Stream
+      .bracket(acquireConnection) {
+        case RabbitConnection(conn) =>
+          L.info(s"Releasing connection: $conn previously acquired.") *>
+            F.delay { if (conn.isOpen) conn.close() }
+      }
+
   /**
     * Creates a connection and a channel in a safe way using Stream.bracket.
     * In case of failure, the resources will be cleaned up properly.
     **/
   override def createConnectionChannel: Stream[F, AMQPChannel] =
     Stream
-      .bracket(acquireConnection) {
+      .bracket(acquireConnectionChannel) {
         case (conn, RabbitChannel(channel)) =>
           L.info(s"Releasing connection: $conn previously acquired.") *>
             F.delay { if (channel.isOpen) channel.close() } *> F.delay { if (conn.isOpen) conn.close() }

core/src/main/scala/com/github/gvolpe/fs2rabbit/interpreter/Fs2Rabbit.scala

@@ -16,7 +16,6 @@
 
 package com.github.gvolpe.fs2rabbit.interpreter
 
-import javax.net.ssl.SSLContext
 import cats.effect.{Concurrent, ConcurrentEffect}
 import cats.syntax.functor._
 import com.github.gvolpe.fs2rabbit.algebra._
@@ -27,6 +26,7 @@ import com.github.gvolpe.fs2rabbit.effects.{EnvelopeDecoder, MessageEncoder}
 import com.github.gvolpe.fs2rabbit.model._
 import com.github.gvolpe.fs2rabbit.program._
 import fs2.Stream
+import javax.net.ssl.SSLContext
 
 // $COVERAGE-OFF$
 object Fs2Rabbit {
@@ -60,6 +60,8 @@ class Fs2Rabbit[F[_]: Concurrent] private[fs2rabbit] (
   private[fs2rabbit] val publishingProgram: Publishing[Stream[F, ?], F] =
     new PublishingProgram[F](amqpClient)
 
+  def createConnection: Stream[F, AMQPConnection] = connectionStream.createConnection
+
   def createConnectionChannel: Stream[F, AMQPChannel] = connectionStream.createConnectionChannel
 
   def createAckerConsumer[A](

core/src/main/scala/com/github/gvolpe/fs2rabbit/model.scala

@@ -26,7 +26,7 @@ import com.github.gvolpe.fs2rabbit.arguments.Arguments
 import com.github.gvolpe.fs2rabbit.effects.{EnvelopeDecoder, MessageEncoder}
 import com.github.gvolpe.fs2rabbit.model.AmqpHeaderVal._
 import com.rabbitmq.client.impl.LongStringHelper
-import com.rabbitmq.client.{AMQP, Channel, LongString}
+import com.rabbitmq.client.{AMQP, Channel, Connection, LongString}
 import fs2.Stream
 
 import scala.collection.JavaConverters._
@@ -40,6 +40,11 @@ object model {
   }
   case class RabbitChannel(value: Channel) extends AMQPChannel
 
+  trait AMQPConnection {
+    def value: Connection
+  }
+  case class RabbitConnection(value: Connection) extends AMQPConnection
+
   case class ExchangeName(value: String) extends AnyVal
   case class QueueName(value: String)    extends AnyVal
   case class RoutingKey(value: String)   extends AnyVal

version.sbt

@@ -1,2 +1,2 @@
-version in ThisBuild := "1.2.0"
+version in ThisBuild := "1.2.1-SNAPSHOT"