Gavin Bisesi Daenyth

Unordered means that we don't care about the order of results from the evalMap action.

It allows for higher throughput because it never waits to start a new job. It has N job permits (with a semaphore), and the moment one job finishes, it requests the next element from the stream and begins operation.

When you use parEvalMap with ordered results, it means that it only begins the next job if the oldest input's job is ready to emit.

This matters when individual elements can take a variable amount of time to complete - and that's the case here, because backfill can take more or less time depending on how many transactions are present within the time window.

Suppose we have 4 jobs we want to run with up to 2 at a time. Job 1 takes 60 seconds to complete, and all the rest take 10 seconds. Using parEvalMap would mean the entire set of inputs would take ~70 seconds to complete.

	#!/usr/bin/env bash

	# Exit immediately if a command exits with a non-zero status.
	set -e
	# Treat unset variables as an error and exit immediately.
	set -u
	# Return the exit status of the last command in the pipeline that failed.
	set -o pipefail

	PROJECT_ID="MYPROJECT"

	//> using dep "com.typesafe:config:1.4.3"
	//> using dep "com.github.pureconfig::pureconfig:0.17.6"
	//> using scala "2.13.14"

	import com.typesafe.config._
	import pureconfig.generic.semiauto._
	import pureconfig.ConfigReader
	import pureconfig.ConfigSource

	import java.{util => ju}

	Test / scalacOptions ++= Seq(
	// Allow using -Wnonunit-statement to find bugs in tests without exploding from scalatest assertions
	"-Wconf:msg=unused value of type org.scalatest.Assertion:s",
	"-Wconf:msg=unused value of type org.scalamock:s"
	)

	/** Execute `onTimeout` every time the stream goes `timeout` duration with no
	* elements
	*/
	def resetTimeout[F[_]: Temporal, A](
	timeout: FiniteDuration,
	onTimeout: F[Unit]
	): fs2.Pipe[F, A, A] = {
	def go(timedPull: Pull.Timed[F, A]): Pull[F, A, Unit] =
	timedPull.timeout(timeout) >>
	timedPull.uncons.flatMap {

	package myapp.users
	// This is the shape of code I've found to work best for doobie in a cats-effect application using capability traits

	trait UserOnboarding[F[_]] {
	// The api expresses the business-level workflow, not the implementation of inserting to a database
	// Keep the interface expressing the high level concern; database is only one implementation and it can change later
	def registerUser(userInfo: UserInfo): F[UserId]
	}

	import cats.{Contravariant, Functor, Semigroupal}
	import slick.jdbc.{
	GetResult,
	PositionedParameters,
	PositionedResult,
	SetParameter
	}


	trait RawSqlInstances {

	# ~/.ssh/config
	Match host github.com exec "[[ $(git remote get-url origin \| sed 's/^.://' \| sed 's/\/.//') = TheGithubOrganization ]]"
	# Don't forget to ssh-add this in addition to other keys, or git will pick one that's already in the agent first
	IdentityFile ~/.ssh/specific_rsa

	import cats.kernel.laws.discipline.SemigroupTests
	import cats.syntax.all._
	import munit.DisciplineSuite
	import org.scalacheck.Arbitrary

	case class MyInt(value: Int)
	object MyInt {
	implicit val semigroupMyInt: Semigroup[MyInt] = Semigroup.instance((x1, x2) => MyInt(x1.value + x2.value))
	}