#593

Build # #593

Build Type

push

Committed by

Commit Message

<a href="https://github.com/evolution-gaming/skafka/commit/<a class=hub.com/evolution-gaming/skafka/commit/46a2b419ef28249a29f089d609a7992830f733ae">46a2b419e<a href="https://github.com/evolution-gaming/skafka/commit/46a2b419ef28249a29f089d609a7992830f733ae">">Merge </a><a class="double-link" href="https://github.com/evolution-gaming/skafka/commit/<a class="double-link" href="https://github.com/evolution-gaming/skafka/commit/61ff053cd93043789758fc2e1cc9692e2b55d5bb">61ff053cd</a>">61ff053cd</a><a href="https://github.com/evolution-gaming/skafka/commit/46a2b419ef28249a29f089d609a7992830f733ae"> into b76b257d5">b76b257d5</a>

Run Details

122 of 122 new or added lines in 6 files covered. (100.0%)

1053 of 1232 relevant lines covered (85.47%)

0.85 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.63

/skafka/src/main/scala/com/evolutiongaming/skafka/Converters.scala

package com.evolutiongaming.skafka

import java.lang.{Long => LongJ}
import java.time.{Duration => DurationJ}
import java.util.{Optional, Collection => CollectionJ, Map => MapJ, Set => SetJ, List => ListJ}

import cats.Monad
import cats.data.{NonEmptyList => Nel, NonEmptySet => Nes, NonEmptyMap => Nem}
import cats.implicits._
import com.evolutiongaming.catshelper.CatsHelper._
import com.evolutiongaming.catshelper.{ApplicativeThrowable, FromTry, MonadThrowable, ToTry}
import org.apache.kafka.clients.consumer.{OffsetAndMetadata => OffsetAndMetadataJ}
import org.apache.kafka.common.header.{Header => HeaderJ}
import org.apache.kafka.common.serialization.{Deserializer, Serializer}
import org.apache.kafka.common.{PartitionInfo => PartitionInfoJ, TopicPartition => TopicPartitionJ}

import scala.compat.java8.DurationConverters
import scala.concurrent.duration.FiniteDuration
import scala.jdk.CollectionConverters._

object Converters {

  implicit class HeaderOps(val self: Header) extends AnyVal {

    def asJava: HeaderJ = new HeaderJ {
      def value = self.value

      def key = self.key
    }
  }


  implicit class HeaderJOps(val self: HeaderJ) extends AnyVal {
    def asScala: Header = Header(self.key(), self.value())
  }


  implicit class AnyOps[A](val self: A) extends AnyVal {
    def noneIf(x: A): Option[A] = if (self == x) None else Some(self)
  }


  implicit class SetTopicPartitionOps(val self: Nes[TopicPartition]) extends AnyVal {

    def asJava: SetJ[TopicPartitionJ] = {
      self
        .toSortedSet
        .toSet[TopicPartition]
        .map { _.asJava }
        .asJava
    }
  }


  implicit class TopicPartitionOps(val self: TopicPartition) extends AnyVal {
    def asJava: TopicPartitionJ = new TopicPartitionJ(self.topic, self.partition.value)
  }


  implicit class TopicPartitionJOps(val self: TopicPartitionJ) extends AnyVal {

    def asScala[F[_] : ApplicativeThrowable]: F[TopicPartition] = {
      for {
        partition <- Partition.of[F](self.partition())
      } yield {
        TopicPartition(self.topic(), partition)
      }
    }
  }


  implicit class PartitionInfoJOps(val self: PartitionInfoJ) extends AnyVal {

    def asScala[F[_] : ApplicativeThrowable]: F[PartitionInfo] = {
      for {
        partition <- Partition.of[F](self.partition())
      } yield {
        PartitionInfo(
          topicPartition = TopicPartition(self.topic, partition),
          leader = self.leader,
          replicas = self.replicas.toList,
          inSyncReplicas = self.inSyncReplicas.toList,
          offlineReplicas = self.offlineReplicas.toList)
      }
    }
  }

  implicit class PartitionInfoOps(val self: PartitionInfo) extends AnyVal {

    def asJava: PartitionInfoJ = {
      new PartitionInfoJ(
        self.topic,
        self.partition.value,
        self.leader,
        self.replicas.toArray,
        self.inSyncReplicas.toArray,
        self.offlineReplicas.toArray)
    }
  }


  implicit class MapJOps[K, V](val self: MapJ[K, V]) extends AnyVal {

    def asScalaMap[F[_] : Monad, A, B](ka: K => F[A], vb: V => F[B]): F[Map[A, B]] = {
      self
        .asScala
        .toList
        .collect { case (k, v) if k != null && v != null =>
          for {
            a <- ka(k)
            b <- vb(v)
          } yield (a, b)
        }
        .sequence
        .map { _.toMap }
    }
  }


  implicit class MapOps[K, V](val self: Map[K, V]) extends AnyVal {

    def asJavaMap[KK, VV](kf: K => KK, vf: V => VV): MapJ[KK, VV] = {
      val zero = new java.util.HashMap[KK, VV]()
      self.foldLeft(zero) { case (map, (k, v)) =>
        val kk = kf(k)
        val vv = vf(v)
        map.put(kk, vv)
        map
      }
    }
  }


  implicit class SerializerOps[A](val self: Serializer[A]) extends AnyVal {

    def asScala[F[_] : FromTry]: ToBytes[F, A] = (a: A, topic: Topic) => {
      FromTry[F].unsafe { self.serialize(topic, a) }
    }
  }


  implicit class DeserializerOps[A](val self: Deserializer[A]) extends AnyVal {

    def asScala[F[_] : FromTry]: FromBytes[F, A] = (value: Bytes, topic: Topic) => {
      FromTry[F].unsafe { self.deserialize(topic, value) }
    }
  }


  implicit class ToBytesOps[F[_], A](val self: ToBytes[F, A]) extends AnyVal {

    def asJava(implicit toTry: ToTry[F]): Serializer[A] = new Serializer[A] {
      override def configure(configs: MapJ[String, _], isKey: Boolean): Unit = {}

      def serialize(topic: Topic, a: A): Array[Byte] = self(a, topic).toTry.get

      override def close() = {}
    }
  }


  implicit class SkafkaFromBytesOps[F[_], A](val self: FromBytes[F, A]) extends AnyVal {

    def asJava(implicit toTry: ToTry[F]): Deserializer[A] = new Deserializer[A] {
      override def configure(configs: MapJ[String, _], isKey: Boolean) = {}

      def deserialize(topic: Topic, bytes: Array[Byte]): A = self(bytes, topic).toTry.get

      override def close() = {}
    }
  }


  implicit class SkafkaNelOps[A](val self: Nel[A]) extends AnyVal {
    def asJava: CollectionJ[A] = self.toList.asJavaCollection
  }


  implicit final class SkafkaDurationJOps(val duration: DurationJ) extends AnyVal {

    def asScala: scala.concurrent.duration.FiniteDuration = DurationConverters.toScala(duration)
  }


  implicit final class SkafkaFiniteDurationOps(val duration: FiniteDuration) extends AnyVal {

    def asJava: java.time.Duration = DurationConverters.toJava(duration)
  }


  implicit class SkafkaOffsetAndMetadataJOpsConverters(val self: OffsetAndMetadataJ) extends AnyVal {

    def asScala[F[_] : ApplicativeThrowable]: F[OffsetAndMetadata] = {
      for {
        offset <- Offset.of[F](self.offset())
      } yield {
        OffsetAndMetadata(offset, self.metadata())
      }
    }
  }


  implicit class SkafkaOffsetAndMetadataOpsConverters(val self: OffsetAndMetadata) extends AnyVal {

    def asJava: OffsetAndMetadataJ = new OffsetAndMetadataJ(self.offset.value, self.metadata)
  }


  implicit class OptionalOpsConverters[A](val self: Optional[A]) extends AnyVal {

    def toOption: Option[A] = if (self.isPresent) self.get().some else none
  }


  implicit class OptionOpsConverters[A](val self: Option[A]) extends AnyVal {

    def toOptional: Optional[A] = self.fold(Optional.empty[A]()) { a => Optional.of(a) }
  }

  def committedOffsetsF[F[_] : MonadThrowable](mapJ: MapJ[TopicPartitionJ, OffsetAndMetadataJ]): F[Map[TopicPartition, OffsetAndMetadata]] = {
    Option(mapJ).fold {
      Map.empty[TopicPartition, OffsetAndMetadata].pure[F]
    } {
      _.asScalaMap(_.asScala[F], _.asScala[F])
    }
  }

  def offsetsMapF[F[_] : MonadThrowable](mapJ: MapJ[TopicPartitionJ, LongJ]): F[Map[TopicPartition, Offset]] = {
    mapJ.asScalaMap(_.asScala[F], a => Offset.of[F](a))
  }

  def asOffsetsJ(offsets: Nem[TopicPartition, OffsetAndMetadata]): MapJ[TopicPartitionJ, OffsetAndMetadataJ] = {
    offsets.toSortedMap.asJavaMap(_.asJava, _.asJava)
  }

  def partitionsInfoListF[F[_]: ApplicativeThrowable](listJ: ListJ[PartitionInfoJ]): F[List[PartitionInfo]] = {
    listJ.asScala.toList.traverse { _.asScala[F] }
  }

  def partitionsInfoMapF[F[_]: MonadThrowable](mapJ: MapJ[Topic, ListJ[PartitionInfoJ]]): F[Map[Topic, List[PartitionInfo]]] = {
    mapJ.asScalaMap(_.pure[F], partitionsInfoListF[F])
  }

  def topicPartitionsSetF[F[_] : ApplicativeThrowable](setJ: SetJ[TopicPartitionJ]): F[Set[TopicPartition]] = {
    for {
      r <- setJ.asScala.toList.traverse { _.asScala[F] }
    } yield r.toSet
  }
}

1	package com.evolutiongaming.skafka
2
3	import java.lang.{Long => LongJ}
4	import java.time.{Duration => DurationJ}
5	import java.util.{Optional, Collection => CollectionJ, Map => MapJ, Set => SetJ, List => ListJ}
6
7	import cats.Monad
8	import cats.data.{NonEmptyList => Nel, NonEmptySet => Nes, NonEmptyMap => Nem}
9	import cats.implicits._
10	import com.evolutiongaming.catshelper.CatsHelper._
11	import com.evolutiongaming.catshelper.{ApplicativeThrowable, FromTry, MonadThrowable, ToTry}
12	import org.apache.kafka.clients.consumer.{OffsetAndMetadata => OffsetAndMetadataJ}
13	import org.apache.kafka.common.header.{Header => HeaderJ}
14	import org.apache.kafka.common.serialization.{Deserializer, Serializer}
15	import org.apache.kafka.common.{PartitionInfo => PartitionInfoJ, TopicPartition => TopicPartitionJ}
16
17	import scala.compat.java8.DurationConverters
18	import scala.concurrent.duration.FiniteDuration
19	import scala.jdk.CollectionConverters._
20
21	object Converters {
22
23	implicit class HeaderOps(val self: Header) extends AnyVal {
24
25	def asJava: HeaderJ = new HeaderJ {	1✔
26	def value = self.value	1✔
27
28	def key = self.key	1✔
29	}
30	}
31
32
33	implicit class HeaderJOps(val self: HeaderJ) extends AnyVal {
34	def asScala: Header = Header(self.key(), self.value())	1✔
35	}
36
37
38	implicit class AnyOps[A](val self: A) extends AnyVal {
39	def noneIf(x: A): Option[A] = if (self == x) None else Some(self)	1✔
40	}
41
42
43	implicit class SetTopicPartitionOps(val self: Nes[TopicPartition]) extends AnyVal {
44
45	def asJava: SetJ[TopicPartitionJ] = {
46	self	1✔
47	.toSortedSet
48	.toSet[TopicPartition]
49	.map { _.asJava }	1✔
50	.asJava	1✔
51	}
52	}
53
54
55	implicit class TopicPartitionOps(val self: TopicPartition) extends AnyVal {
56	def asJava: TopicPartitionJ = new TopicPartitionJ(self.topic, self.partition.value)	1✔
57	}
58
59
60	implicit class TopicPartitionJOps(val self: TopicPartitionJ) extends AnyVal {
61
62	def asScala[F[_] : ApplicativeThrowable]: F[TopicPartition] = {
63	for {
64	partition <- Partition.of[F](self.partition())	1✔
65	} yield {
66	TopicPartition(self.topic(), partition)	1✔
67	}
68	}
69	}
70
71
72	implicit class PartitionInfoJOps(val self: PartitionInfoJ) extends AnyVal {
73
74	def asScala[F[_] : ApplicativeThrowable]: F[PartitionInfo] = {
75	for {
76	partition <- Partition.of[F](self.partition())	1✔
77	} yield {
78	PartitionInfo(	1✔
79	topicPartition = TopicPartition(self.topic, partition),	1✔
80	leader = self.leader,	1✔
81	replicas = self.replicas.toList,	1✔
82	inSyncReplicas = self.inSyncReplicas.toList,	1✔
83	offlineReplicas = self.offlineReplicas.toList)	1✔
84	}
85	}
86	}
87
88	implicit class PartitionInfoOps(val self: PartitionInfo) extends AnyVal {
89
90	def asJava: PartitionInfoJ = {
91	new PartitionInfoJ(	1✔
92	self.topic,	1✔
93	self.partition.value,	1✔
94	self.leader,	1✔
95	self.replicas.toArray,	1✔
96	self.inSyncReplicas.toArray,	1✔
97	self.offlineReplicas.toArray)	1✔
98	}
99	}
100
101
102	implicit class MapJOps[K, V](val self: MapJ[K, V]) extends AnyVal {
103
104	def asScalaMap[F[_] : Monad, A, B](ka: K => F[A], vb: V => F[B]): F[Map[A, B]] = {
105	self	1✔
106	.asScala
107	.toList
108	.collect { case (k, v) if k != null && v != null =>	1✔
109	for {
110	a <- ka(k)	1✔
111	b <- vb(v)	1✔
112	} yield (a, b)	1✔
113	}
114	.sequence	1✔
115	.map { _.toMap }	1✔
116	}
117	}
118
119
120	implicit class MapOps[K, V](val self: Map[K, V]) extends AnyVal {
121
122	def asJavaMap[KK, VV](kf: K => KK, vf: V => VV): MapJ[KK, VV] = {
123	val zero = new java.util.HashMap[KK, VV]()	1✔
124	self.foldLeft(zero) { case (map, (k, v)) =>	1✔
125	val kk = kf(k)	1✔
126	val vv = vf(v)	1✔
127	map.put(kk, vv)	1✔
128	map
129	}
130	}
131	}
132
133
134	implicit class SerializerOps[A](val self: Serializer[A]) extends AnyVal {
135
136	def asScala[F[_] : FromTry]: ToBytes[F, A] = (a: A, topic: Topic) => {
137	FromTry[F].unsafe { self.serialize(topic, a) }	1✔
138	}
139	}
140
141
142	implicit class DeserializerOps[A](val self: Deserializer[A]) extends AnyVal {
143
144	def asScala[F[_] : FromTry]: FromBytes[F, A] = (value: Bytes, topic: Topic) => {
145	FromTry[F].unsafe { self.deserialize(topic, value) }	1✔
146	}
147	}
148
149
150	implicit class ToBytesOps[F[_], A](val self: ToBytes[F, A]) extends AnyVal {
151
152	def asJava(implicit toTry: ToTry[F]): Serializer[A] = new Serializer[A] {	1✔
153	override def configure(configs: MapJ[String, _], isKey: Boolean): Unit = {}	×
154
155	def serialize(topic: Topic, a: A): Array[Byte] = self(a, topic).toTry.get	1✔
156
157	override def close() = {}	×
158	}
159	}
160
161
162	implicit class SkafkaFromBytesOps[F[_], A](val self: FromBytes[F, A]) extends AnyVal {
163
164	def asJava(implicit toTry: ToTry[F]): Deserializer[A] = new Deserializer[A] {	1✔
165	override def configure(configs: MapJ[String, _], isKey: Boolean) = {}	×
166
167	def deserialize(topic: Topic, bytes: Array[Byte]): A = self(bytes, topic).toTry.get	1✔
168
169	override def close() = {}	1✔
170	}
171	}
172
173
174	implicit class SkafkaNelOps[A](val self: Nel[A]) extends AnyVal {
175	def asJava: CollectionJ[A] = self.toList.asJavaCollection	×
176	}
177
178
179	implicit final class SkafkaDurationJOps(val duration: DurationJ) extends AnyVal {
180
181	def asScala: scala.concurrent.duration.FiniteDuration = DurationConverters.toScala(duration)	1✔
182	}
183
184
185	implicit final class SkafkaFiniteDurationOps(val duration: FiniteDuration) extends AnyVal {
186
187	def asJava: java.time.Duration = DurationConverters.toJava(duration)	1✔
188	}
189
190
191	implicit class SkafkaOffsetAndMetadataJOpsConverters(val self: OffsetAndMetadataJ) extends AnyVal {
192
193	def asScala[F[_] : ApplicativeThrowable]: F[OffsetAndMetadata] = {
194	for {
195	offset <- Offset.of[F](self.offset())	1✔
196	} yield {
197	OffsetAndMetadata(offset, self.metadata())	1✔
198	}
199	}
200	}
201
202
203	implicit class SkafkaOffsetAndMetadataOpsConverters(val self: OffsetAndMetadata) extends AnyVal {
204
205	def asJava: OffsetAndMetadataJ = new OffsetAndMetadataJ(self.offset.value, self.metadata)	1✔
206	}
207
208
209	implicit class OptionalOpsConverters[A](val self: Optional[A]) extends AnyVal {
210
211	def toOption: Option[A] = if (self.isPresent) self.get().some else none	1✔
212	}
213
214
215	implicit class OptionOpsConverters[A](val self: Option[A]) extends AnyVal {
216
217	def toOptional: Optional[A] = self.fold(Optional.empty[A]()) { a => Optional.of(a) }	×
218	}
219
220	def committedOffsetsF[F[_] : MonadThrowable](mapJ: MapJ[TopicPartitionJ, OffsetAndMetadataJ]): F[Map[TopicPartition, OffsetAndMetadata]] = {
221	Option(mapJ).fold {
222	Map.empty[TopicPartition, OffsetAndMetadata].pure[F]	×
223	} {	1✔
224	_.asScalaMap(_.asScala[F], _.asScala[F])	1✔
225	}
226	}
227
228	def offsetsMapF[F[_] : MonadThrowable](mapJ: MapJ[TopicPartitionJ, LongJ]): F[Map[TopicPartition, Offset]] = {
229	mapJ.asScalaMap(_.asScala[F], a => Offset.of[F](a))	1✔
230	}
231
232	def asOffsetsJ(offsets: Nem[TopicPartition, OffsetAndMetadata]): MapJ[TopicPartitionJ, OffsetAndMetadataJ] = {
233	offsets.toSortedMap.asJavaMap(_.asJava, _.asJava)	1✔
234	}
235
236	def partitionsInfoListF[F[_]: ApplicativeThrowable](listJ: ListJ[PartitionInfoJ]): F[List[PartitionInfo]] = {
237	listJ.asScala.toList.traverse { _.asScala[F] }	1✔
238	}
239
240	def partitionsInfoMapF[F[_]: MonadThrowable](mapJ: MapJ[Topic, ListJ[PartitionInfoJ]]): F[Map[Topic, List[PartitionInfo]]] = {
241	mapJ.asScalaMap(_.pure[F], partitionsInfoListF[F])	1✔
242	}
243
244	def topicPartitionsSetF[F[_] : ApplicativeThrowable](setJ: SetJ[TopicPartitionJ]): F[Set[TopicPartition]] = {
245	for {
246	r <- setJ.asScala.toList.traverse { _.asScala[F] }	1✔
247	} yield r.toSet	1✔
248	}
249	}

evolution-gaming / skafka / #593

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous