2080

pending completion

Build # 2080

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/outworkers/phantom/commit/<a class=hub.com/outworkers/phantom/commit/<a class="double-link" href="https://git"><a class=hub.com/outworkers/phantom/commit/<a class="double-link" href="https://git"><a class=hub.com/outworkers/phantom/commit/<a class="double-link" href="https://git"><a class=hub.com/outworkers/phantom/commit/<a class="double-link" href="https://git"><a class=hub.com/outworkers/phantom/commit/5b8969e77c8db69b407df9cdc30136164a11158d">5b8969e77">&lt;a href=&quot;https://github.com/outworkers/phantom/commit/</a><a class="double-link" href="https://github.com/outworkers/phantom/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/outworkers/phantom/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/outworkers/phantom/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/outworkers/phantom/commit/5b8969e77c8db69b407df9cdc30136164a11158d">5b8969e77</a><a href="https://github.com/outworkers/phantom/commit/5b8969e77c8db69b407df9cdc30136164a11158d">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/outworkers/phantom/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/outworkers/phantom/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/outworkers/phantom/commit/5b8969e77c8db69b407df9cdc30136164a11158d&quot;&gt;5b8969e77&lt;/a&gt;&lt;a href=&quot;https://github.com/outworkers/phantom/commit/5b8969e77c8db69b407df9cdc30136164a11158d&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/outworkers/phantom/commit/5b8969e77c8db69b407df9cdc30136164a11158d&amp;quot;&amp;gt;&amp;amp;quot;&amp;amp;gt;Feature/patching build (#840)

* Adding support for multi arg aggregate calls

* Using tuple implicits

* Adding ~ combinators for multi-argument selectors

* Adding multi aggregate functions.

* Spacing things out

* Adding an HList extractor for nested options

* removing some more unused imports

* Adding logging for when routing key functions are applied

* Correcting some typos

* Adding documentation for the cluster builder

* Better docs

* oops

* fixing some unused imports and fixing tests

* Oops

* Removing the need to have a Tuple1 extractor manually defined wi... (continued)

Run Details

2912 of 3373 relevant lines covered (86.33%)

0.86 hits per line

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

80.0

/phantom-streams/src/main/scala/com/outworkers/phantom/streams/iteratee/Enumerator.scala

/*
 * Copyright 2013 - 2017 Outworkers Ltd.
 *
 * 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.outworkers.phantom.streams.iteratee

import com.outworkers.phantom.{ResultSet, Row}
import play.api.libs.iteratee.Execution.{defaultExecutionContext => dec}
import play.api.libs.iteratee.{Cont, Done, Error, Input, Step, Enumerator => PlayEnum, Iteratee => PlayIter}

import scala.concurrent.Future

private[phantom] object Enumerator {

  def enumerator(resultSet: ResultSet)(implicit ctx: scala.concurrent.ExecutionContext): PlayEnum[Row] = {

    new PlayEnum[Row] {
      private[this] val rs = resultSet
      private[this] val it = rs.iterate()

      def apply[A](iter: PlayIter[Row, A]): Future[PlayIter[Row, A]] = {

        def step(iter: PlayIter[Row, A]): Future[PlayIter[Row, A]] = {
          iter.fold {
            case Step.Cont(k) => fetch[A](step, k)
            case Step.Done(a, e) => Future.successful(Done(a, e))
            case Step.Error(msg, inp) => Future.successful(Error(msg, inp))
          }(dec)
        }

        step(iter)
      }

      def fetch[A](
        loop: PlayIter[Row, A] => Future[PlayIter[Row, A]],
        k: Input[Row] => PlayIter[Row, A]
      ): Future[PlayIter[Row, A]] = {
        val available = resultSet.getAvailableWithoutFetching

        if (!rs.isExhausted) {
          // prefetch if we are running low on results.
          if (available < 100 && !resultSet.isFullyFetched) {
            rs.fetchMoreResults
          }
          // if we have less than one result available non-blocking
          // we send the next invocation of the iterator to the thread
          // pool as blocking
          if (available < 1) {
            Future(it.next)(ctx).flatMap(row => loop(k(Input.El(row))))(dec)
          }
          // otherwise we can just carry on.
          else {
            loop(k(Input.El(it.next)))
          }
        } else {
          Future.successful(Cont(k))
        }
      }
    }
  }
}


1	/*
2	* Copyright 2013 - 2017 Outworkers Ltd.
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	package com.outworkers.phantom.streams.iteratee
17
18	import com.outworkers.phantom.{ResultSet, Row}
19	import play.api.libs.iteratee.Execution.{defaultExecutionContext => dec}
20	import play.api.libs.iteratee.{Cont, Done, Error, Input, Step, Enumerator => PlayEnum, Iteratee => PlayIter}
21
22	import scala.concurrent.Future
23
24	private[phantom] object Enumerator {
25
26	def enumerator(resultSet: ResultSet)(implicit ctx: scala.concurrent.ExecutionContext): PlayEnum[Row] = {
27
28	new PlayEnum[Row] {	1✔
29	private[this] val rs = resultSet
30	private[this] val it = rs.iterate()	1✔
31
32	def apply[A](iter: PlayIter[Row, A]): Future[PlayIter[Row, A]] = {
33
34	def step(iter: PlayIter[Row, A]): Future[PlayIter[Row, A]] = {
35	iter.fold {
36	case Step.Cont(k) => fetch[A](step, k)	1✔
37	case Step.Done(a, e) => Future.successful(Done(a, e))	×
38	case Step.Error(msg, inp) => Future.successful(Error(msg, inp))	×
39	}(dec)	1✔
40	}
41
42	step(iter)	1✔
43	}
44
45	def fetch[A](
46	loop: PlayIter[Row, A] => Future[PlayIter[Row, A]],
47	k: Input[Row] => PlayIter[Row, A]
48	): Future[PlayIter[Row, A]] = {
49	val available = resultSet.getAvailableWithoutFetching	1✔
50
51	if (!rs.isExhausted) {	1✔
52	// prefetch if we are running low on results.
53	if (available < 100 && !resultSet.isFullyFetched) {	1✔
54	rs.fetchMoreResults	1✔
55	}
56	// if we have less than one result available non-blocking
57	// we send the next invocation of the iterator to the thread
58	// pool as blocking
59	if (available < 1) {	1✔
60	Future(it.next)(ctx).flatMap(row => loop(k(Input.El(row))))(dec)	×
61	}
62	// otherwise we can just carry on.
63	else {
64	loop(k(Input.El(it.next)))	1✔
65	}
66	} else {
67	Future.successful(Cont(k))	1✔
68	}
69	}
70	}
71	}
72	}
73

outworkers / phantom / 2080

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