1+ /*
2+ * Copyright (c) 2014-2021 by The Monix Project Developers.
3+ *
4+ * Licensed under the Apache License, Version 2.0 (the "License");
5+ * you may not use this file except in compliance with the License.
6+ * You may obtain a copy of the License at
7+ *
8+ * http://www.apache.org/licenses/LICENSE-2.0
9+ *
10+ * Unless required by applicable law or agreed to in writing, software
11+ * distributed under the License is distributed on an "AS IS" BASIS,
12+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+ * See the License for the specific language governing permissions and
14+ * limitations under the License.
15+ */
16+
17+ package monix .kafka
18+
19+ import cats .effect .Resource
20+ import monix .eval .Task
21+ import monix .kafka .KafkaConsumerObservable .createConsumer
22+ import monix .kafka .config .ObservableCommitOrder
23+ import org .apache .kafka .clients .consumer .{Consumer , ConsumerRecord , KafkaConsumer }
24+
25+ import scala .concurrent .blocking
26+ import scala .util .matching .Regex
27+
28+ /** Exposes an `Observable` that consumes a Kafka stream by
29+ * means of a Kafka Consumer client.
30+ *
31+ * In order to get initialized, it needs a configuration. See the
32+ * [[KafkaConsumerConfig ]] needed and see `monix/kafka/default.conf`,
33+ * (in the resource files) that is exposing all default values.
34+ */
35+ object KafkaConsumerResource {
36+
37+
38+ /** A [[Resource ]] that acquires a [[KafkaConsumer ]] used
39+ * to build a [[KafkaConsumerObservableAutoCommit ]] instance,
40+ * that will be released after it's usage.
41+ *
42+ * @note The consumer will act consequently depending on the
43+ * [[ObservableCommitOrder ]] that was chosen from configuration.
44+ * Which can be configured from the key `monix.observable.commit.order`.
45+ *
46+ * @param cfg is the [[KafkaConsumerConfig ]] needed for initializing the
47+ * consumer; also make sure to see `monix/kafka/default.conf` for
48+ * the default values being used.
49+ * @param consumer is a factory for the
50+ * `org.apache.kafka.clients.consumer.KafkaConsumer`
51+ * instance to use for consuming from Kafka
52+ */
53+ def apply [K , V ](
54+ cfg : KafkaConsumerConfig ,
55+ consumer : Task [Consumer [K , V ]]): Resource [Task , KafkaConsumerObservable [K , V , ConsumerRecord [K , V ]]]= {
56+ for {
57+ consumer <- Resource .make(consumer){ consumer =>
58+ Task .evalAsync(consumer.synchronized { blocking(consumer.close())})
59+ }
60+ consumerObservable <- Resource .liftF(Task (new KafkaConsumerObservableAutoCommit [K , V ](cfg, Task .pure(consumer), shouldClose = false )))
61+ } yield consumerObservable
62+ }
63+
64+ /** A [[Resource ]] that acquires a [[KafkaConsumer ]] used
65+ * to build a [[KafkaConsumerObservableAutoCommit ]] instance,
66+ * that will be released after it's usage.
67+ *
68+ * @note The consumer will act consequently depending on the
69+ * [[ObservableCommitOrder ]] that was chosen from configuration.
70+ * Which can be configured from the key `monix.observable.commit.order`.
71+ *
72+ * @param cfg is the [[KafkaConsumerConfig ]] needed for initializing the
73+ * consumer; also make sure to see `monix/kafka/default.conf` for
74+ * the default values being used.
75+ * @param topics is the list of Kafka topics to subscribe to.
76+ */
77+ def apply [K , V ](cfg : KafkaConsumerConfig , topics : List [String ])(implicit
78+ K : Deserializer [K ],
79+ V : Deserializer [V ]): Resource [Task , KafkaConsumerObservable [K , V , ConsumerRecord [K , V ]]] = {
80+ val consumer = createConsumer[K , V ](cfg, topics)
81+ apply(cfg, consumer)
82+ }
83+
84+ /** A [[Resource ]] that acquires a [[KafkaConsumer ]] used
85+ * to build a [[KafkaConsumerObservableAutoCommit ]] instance,
86+ * that will be released after it's usage.
87+ *
88+ * @note The consumer will act consequently depending on the
89+ * [[ObservableCommitOrder ]] that was chosen from configuration.
90+ * Which can be configured from the key `monix.observable.commit.order`.
91+ *
92+ * @param cfg is the [[KafkaConsumerConfig ]] needed for initializing the
93+ * consumer; also make sure to see `monix/kafka/default.conf` for
94+ * the default values being used.
95+ * @param topicsRegex is the pattern of Kafka topics to subscribe to.
96+ */
97+ def apply [K , V ](cfg : KafkaConsumerConfig , topicsRegex : Regex )(implicit
98+ K : Deserializer [K ],
99+ V : Deserializer [V ]): Resource [Task , KafkaConsumerObservable [K , V , ConsumerRecord [K , V ]]] = {
100+
101+ val consumer = createConsumer[K , V ](cfg, topicsRegex)
102+ apply(cfg, consumer)
103+ }
104+
105+ /**
106+ * A [[Resource ]] that acquires a [[KafkaConsumer ]] used
107+ * to build a [[KafkaConsumerObservableManualCommit ]] instance,
108+ * which provides the ability to manual commit offsets and
109+ * forcibly disables auto commits in configuration.
110+ * Such instances emit [[CommittableMessage ]] instead of Kafka's [[ConsumerRecord ]].
111+ *
112+ * ==Example==
113+ * {{{
114+ * KafkaConsumerResource.manualCommit[String,String](consumerCfg, List(topicName)).use{ committableMessages =>
115+ * committableMessages.map(message => message.record.value() -> message.committableOffset)
116+ * .mapEval { case (value, offset) => performBusinessLogic(value).map(_ => offset) }
117+ * .bufferTimedAndCounted(1.second, 1000)
118+ * .mapEval(offsets => CommittableOffsetBatch(offsets).commitAsync())
119+ * .completedL
120+ * }}}
121+ *
122+ * @param cfg is the [[KafkaConsumerConfig ]] needed for initializing the
123+ * consumer; also make sure to see `monix/kafka/default.conf` for
124+ * the default values being used. Auto commit will disabled and
125+ * observable commit order will turned to [[monix.kafka.config.ObservableCommitOrder.NoAck NoAck ]] forcibly!
126+ *
127+ * @param consumer is a factory for the
128+ * `org.apache.kafka.clients.consumer.KafkaConsumer`
129+ * instance to use for consuming from Kafka
130+ */
131+ def manualCommit [K , V ](
132+ cfg : KafkaConsumerConfig ,
133+ consumer : Task [Consumer [K , V ]]): Resource [Task , KafkaConsumerObservable [K , V , CommittableMessage [K , V ]]] = {
134+ for {
135+ consumer <- Resource .make(consumer){ consumer =>
136+ Task .evalAsync(consumer.synchronized { blocking(consumer.close())})
137+ }
138+ manualCommitConf = cfg.copy(observableCommitOrder = ObservableCommitOrder .NoAck , enableAutoCommit = false )
139+ consumerObservable <- Resource .liftF(Task (new KafkaConsumerObservableManualCommit [K , V ](manualCommitConf, Task .pure(consumer), shouldClose = false )))
140+ } yield consumerObservable
141+ }
142+
143+ /** Builds a [[KafkaConsumerObservable ]] instance with ability to manual commit offsets
144+ * and forcibly disables auto commits in configuration.
145+ * Such instances emit [[CommittableMessage ]] instead of Kafka's ConsumerRecord.
146+ *
147+ * ==Example==
148+ * {{{
149+ * KafkaConsumerResource.manualCommit[String,String](consumerCfg, List(topicName)).use{ committableMessages =>
150+ * committableMessages.map(message => message.record.value() -> message.committableOffset)
151+ * .mapEval { case (value, offset) => performBusinessLogic(value).map(_ => offset) }
152+ * .bufferTimedAndCounted(1.second, 1000)
153+ * .mapEval(offsets => CommittableOffsetBatch(offsets).commitSync())
154+ * .completedL
155+ * }
156+ * }}}
157+ *
158+ * @param cfg is the [[KafkaConsumerConfig ]] needed for initializing the
159+ * consumer; also make sure to see `monix/kafka/default.conf` for
160+ * the default values being used. Auto commit will disabled and
161+ * observable commit order will turned to [[monix.kafka.config.ObservableCommitOrder.NoAck NoAck ]] forcibly!
162+ *
163+ * @param topics is the list of Kafka topics to subscribe to.
164+ */
165+ def manualCommit [K , V ](cfg : KafkaConsumerConfig , topics : List [String ])(implicit
166+ K : Deserializer [K ],
167+ V : Deserializer [V ]): Resource [Task , KafkaConsumerObservable [K , V , CommittableMessage [K , V ]]] = {
168+
169+ val consumer = createConsumer[K , V ](cfg, topics)
170+ manualCommit(cfg, consumer)
171+ }
172+
173+ /** Builds a [[KafkaConsumerObservable ]] instance with ability to manual commit offsets
174+ * and forcibly disables auto commits in configuration.
175+ * Such instances emit [[CommittableMessage ]] instead of Kafka's ConsumerRecord.
176+ *
177+ * ==Example==
178+ * {{{
179+ * KafkaConsumerResource.manualCommit[String,String](consumerCfg, List(topicName)).use{ committableMessages =>
180+ * committableMessages.map(message => message.record.value() -> message.committableOffset)
181+ * .mapEval { case (value, offset) => performBusinessLogic(value).map(_ => offset) }
182+ * .bufferTimedAndCounted(1.second, 1000)
183+ * .mapEval(offsets => CommittableOffsetBatch(offsets).commitSync())
184+ * .completedL
185+ * }
186+ * }}}
187+ *
188+ * @param cfg is the [[KafkaConsumerConfig ]] needed for initializing the
189+ * consumer; also make sure to see `monix/kafka/default.conf` for
190+ * the default values being used. Auto commit will disabled and
191+ * observable commit order will turned to [[monix.kafka.config.ObservableCommitOrder.NoAck NoAck ]] forcibly!
192+ *
193+ * @param topicsRegex is the pattern of Kafka topics to subscribe to.
194+ */
195+ def manualCommit [K , V ](cfg : KafkaConsumerConfig , topicsRegex : Regex )(implicit
196+ K : Deserializer [K ],
197+ V : Deserializer [V ]): Resource [Task , KafkaConsumerObservable [K , V , CommittableMessage [K , V ]]] = {
198+
199+ val consumer = createConsumer[K , V ](cfg, topicsRegex)
200+ manualCommit(cfg, consumer)
201+ }
202+
203+ }
0 commit comments