4001

pending completion

Build # 4001

Build Type

Pull #377

travis-ci-com

Committed by

web-flow

Commit Message

<a href="https://github.com/broadinstitute/workbench-libs/commit/<a class=hub.com/broadinstitute/workbench-libs/commit/<a class="double-link" href="https://git"><a class=hub.com/broadinstitute/workbench-libs/commit/<a class="double-link" href="https://git"><a class=hub.com/broadinstitute/workbench-libs/commit/<a class="double-link" href="https://git"><a class=hub.com/broadinstitute/workbench-libs/commit/b31890ddb0c0a8a7b413cf1293b95d6b2a1e6152">b31890ddb">&lt;a href=&quot;https://github.com/broadinstitute/workbench-libs/commit/</a><a class="double-link" href="https://github.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/broadinstitute/workbench-libs/commit/b31890ddb0c0a8a7b413cf1293b95d6b2a1e6152">b31890ddb</a><a href="https://github.com/broadinstitute/workbench-libs/commit/b31890ddb0c0a8a7b413cf1293b95d6b2a1e6152">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/broadinstitute/workbench-libs/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/broadinstitute/workbench-libs/commit/b31890ddb0c0a8a7b413cf1293b95d6b2a1e6152&quot;&gt;b31890ddb&lt;/a&gt;&lt;a href=&quot;https://github.com/broadinstitute/workbench-libs/commit/b31890ddb0c0a8a7b413cf1293b95d6b2a1e6152&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;htt... (continued)

Pull Request Pull Request #377: IA-1790 add list to GoogleTopicAdmin

Run Details

29 of 29 new or added lines in 5 files covered. (100.0%)

393 of 1772 relevant lines covered (22.18%)

0.22 hits per line

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

39.39

/google2/src/main/scala/org/broadinstitute/dsde/workbench/google2/GooglePublisherInterpreter.scala

package org.broadinstitute.dsde.workbench.google2

import cats.effect._
import cats.implicits._
import cats.mtl.Ask
import com.google.api.core.ApiFutures
import com.google.api.gax.core.FixedCredentialsProvider
import com.google.api.gax.rpc.AlreadyExistsException
import com.google.auth.oauth2.ServiceAccountCredentials
import com.google.cloud.pubsub.v1.{Publisher, TopicAdminClient}
import com.google.common.util.concurrent.MoreExecutors
import com.google.protobuf.ByteString
import com.google.pubsub.v1.{ProjectTopicName, PubsubMessage, TopicName}
import fs2.{Pipe, Stream}
import io.chrisdavenport.log4cats.StructuredLogger
import io.circe.Encoder
import io.circe.syntax._
import org.broadinstitute.dsde.workbench.model.TraceId

private[google2] class GooglePublisherInterpreter[F[_]: Async: Timer: StructuredLogger](
  publisher: Publisher
) extends GooglePublisher[F] {
  def publish[MessageType: Encoder]: Pipe[F, MessageType, Unit] = in => {
    in.flatMap { message =>
      Stream
        .eval(publishMessage(message.asJson.noSpaces, None)) //This will turn message case class into raw json string
    }
  }

  def publishNative: Pipe[F, PubsubMessage, Unit] = in => {
    in.flatMap { message =>
      Stream.eval(
        withLogging(
          Async[F]
            .async[String] { callback =>
              ApiFutures.addCallback(
                publisher.publish(message),
                callBack(callback),
                MoreExecutors.directExecutor()
              )
            }
            .void,
          Option(message.getAttributesMap.get("traceId")).map(s => TraceId(s)),
          s"Publishing ${message}"
        )
      )
    }
  }

  def publishString: Pipe[F, String, Unit] = in => {
    in.flatMap(s => Stream.eval(publishMessage(s, None)))
  }

  override def publishOne[MessageType: Encoder](message: MessageType)(implicit ev: Ask[F, TraceId]): F[Unit] = {
    val byteString = ByteString.copyFromUtf8(message.asJson.noSpaces)
    tracedLogging(asyncPublishMessage(byteString), s"com.google.cloud.pubsub.v1.Publisher.publish($byteString)")
  }

  private def publishMessage(message: String, traceId: Option[TraceId]): F[Unit] = {
    val byteString = ByteString.copyFromUtf8(message)
    withLogging(asyncPublishMessage(byteString), traceId, s"Publishing $message")
  }

  private def asyncPublishMessage(byteString: ByteString): F[Unit] =
    Async[F]
      .async[String] { callback =>
        val message = PubsubMessage
          .newBuilder()
          .setData(byteString)
          .build()
        ApiFutures.addCallback(
          publisher.publish(message),
          callBack(callback),
          MoreExecutors.directExecutor()
        )
      }
      .void
}

object GooglePublisherInterpreter {
  def apply[F[_]: Async: Timer: ContextShift: StructuredLogger](
    publisher: Publisher
  ): GooglePublisherInterpreter[F] = new GooglePublisherInterpreter(publisher)

  def publisher[F[_]: Sync: StructuredLogger](config: PublisherConfig): Resource[F, Publisher] =
    for {
      credential <- credentialResource(config.pathToCredentialJson)
      publisher <- publisherResource(config.projectTopicName, credential)
      topicAdminClient <- GoogleTopicAdminInterpreter.topicAdminClientResource(credential)
      _ <- createTopic(config.projectTopicName, topicAdminClient)
    } yield publisher

  private def createTopic[F[_]: Sync](topicName: TopicName, topicAdminClient: TopicAdminClient)(implicit
    logger: StructuredLogger[F]
  ): Resource[F, Unit] =
    Resource.liftF(
      Sync[F]
        .delay(topicAdminClient.createTopic(topicName))
        .void
        .recoverWith { case _: AlreadyExistsException =>
          logger.info(s"${topicName} already exists")
        }
    )

  private def publisherResource[F[_]: Sync](topicName: ProjectTopicName,
                                            credential: ServiceAccountCredentials
  ): Resource[F, Publisher] =
    Resource.make(
      Sync[F].delay(
        Publisher
          .newBuilder(topicName)
          .setCredentialsProvider(FixedCredentialsProvider.create(credential))
          .build()
      )
    )(p => Sync[F].delay(p.shutdown()))
}

final case class PublisherConfig(pathToCredentialJson: String, projectTopicName: ProjectTopicName)

1	package org.broadinstitute.dsde.workbench.google2
2
3	import cats.effect._
4	import cats.implicits._
5	import cats.mtl.Ask
6	import com.google.api.core.ApiFutures
7	import com.google.api.gax.core.FixedCredentialsProvider
8	import com.google.api.gax.rpc.AlreadyExistsException
9	import com.google.auth.oauth2.ServiceAccountCredentials
10	import com.google.cloud.pubsub.v1.{Publisher, TopicAdminClient}
11	import com.google.common.util.concurrent.MoreExecutors
12	import com.google.protobuf.ByteString
13	import com.google.pubsub.v1.{ProjectTopicName, PubsubMessage, TopicName}
14	import fs2.{Pipe, Stream}
15	import io.chrisdavenport.log4cats.StructuredLogger
16	import io.circe.Encoder
17	import io.circe.syntax._
18	import org.broadinstitute.dsde.workbench.model.TraceId
19
20	private[google2] class GooglePublisherInterpreter[F[_]: Async: Timer: StructuredLogger](
21	publisher: Publisher
22	) extends GooglePublisher[F] {
23	def publish[MessageType: Encoder]: Pipe[F, MessageType, Unit] = in => {
24	in.flatMap { message =>	1✔
25	Stream
26	.eval(publishMessage(message.asJson.noSpaces, None)) //This will turn message case class into raw json string	1✔
27	}
28	}
29
30	def publishNative: Pipe[F, PubsubMessage, Unit] = in => {
31	in.flatMap { message =>	×
32	Stream.eval(	×
33	withLogging(	×
34	Async[F]	×
35	.async[String] { callback =>	×
36	ApiFutures.addCallback(	×
37	publisher.publish(message),	×
38	callBack(callback),	×
39	MoreExecutors.directExecutor()	×
40	)
41	}
42	.void,	×
43	Option(message.getAttributesMap.get("traceId")).map(s => TraceId(s)),	×
44	s"Publishing ${message}"
45	)
46	)
47	}
48	}
49
50	def publishString: Pipe[F, String, Unit] = in => {
51	in.flatMap(s => Stream.eval(publishMessage(s, None)))	×
52	}
53
54	override def publishOne[MessageType: Encoder](message: MessageType)(implicit ev: Ask[F, TraceId]): F[Unit] = {
55	val byteString = ByteString.copyFromUtf8(message.asJson.noSpaces)	×
56	tracedLogging(asyncPublishMessage(byteString), s"com.google.cloud.pubsub.v1.Publisher.publish($byteString)")	×
57	}
58
59	private def publishMessage(message: String, traceId: Option[TraceId]): F[Unit] = {
60	val byteString = ByteString.copyFromUtf8(message)	1✔
61	withLogging(asyncPublishMessage(byteString), traceId, s"Publishing $message")	1✔
62	}
63
64	private def asyncPublishMessage(byteString: ByteString): F[Unit] =
65	Async[F]	1✔
66	.async[String] { callback =>	1✔
67	val message = PubsubMessage
68	.newBuilder()
69	.setData(byteString)
70	.build()	1✔
71	ApiFutures.addCallback(	1✔
72	publisher.publish(message),	1✔
73	callBack(callback),	1✔
74	MoreExecutors.directExecutor()	1✔
75	)
76	}
77	.void	1✔
78	}
79
80	object GooglePublisherInterpreter {
81	def apply[F[_]: Async: Timer: ContextShift: StructuredLogger](
82	publisher: Publisher
83	): GooglePublisherInterpreter[F] = new GooglePublisherInterpreter(publisher)	1✔
84
85	def publisher[F[_]: Sync: StructuredLogger](config: PublisherConfig): Resource[F, Publisher] =
86	for {
87	credential <- credentialResource(config.pathToCredentialJson)	×
88	publisher <- publisherResource(config.projectTopicName, credential)	×
89	topicAdminClient <- GoogleTopicAdminInterpreter.topicAdminClientResource(credential)	×
90	_ <- createTopic(config.projectTopicName, topicAdminClient)	×
91	} yield publisher
92
93	private def createTopic[F[_]: Sync](topicName: TopicName, topicAdminClient: TopicAdminClient)(implicit
94	logger: StructuredLogger[F]
95	): Resource[F, Unit] =
96	Resource.liftF(	×
97	Sync[F]
98	.delay(topicAdminClient.createTopic(topicName))
99	.void
100	.recoverWith { case _: AlreadyExistsException =>
101	logger.info(s"${topicName} already exists")
102	}
103	)
104
105	private def publisherResource[F[_]: Sync](topicName: ProjectTopicName,
106	credential: ServiceAccountCredentials
107	): Resource[F, Publisher] =
108	Resource.make(
109	Sync[F].delay(
110	Publisher
111	.newBuilder(topicName)
112	.setCredentialsProvider(FixedCredentialsProvider.create(credential))
113	.build()
114	)
115	)(p => Sync[F].delay(p.shutdown()))	×
116	}
117
118	final case class PublisherConfig(pathToCredentialJson: String, projectTopicName: ProjectTopicName)

broadinstitute / workbench-libs / 4001

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