JohnMurray

class ApiService[T] {
  // other code ...
  
  case class RequestQueueItem(f: () => Future[T],
                              p: Promise[T],
                              limitDetector: LimitDetector,
                              startDeadline: Option[LocalDateTime])
  
}

JohnMurray

class ApiService[T] {
  // other code ...
  
  def request(f: () => Future[T], maxQueueTime: Duration = Duration.Inf)
             (d: LimitDetector = limitDetection)
             : Future[T] = this.synchronized {
    val p = Promise[T]()
    val startDeadline =
      if (maxQueueTime.isFinite)
        Some(LocalDateTime.now().plusSeconds(maxQueueTime.toSeconds))

JohnMurray

class ApiService [T] {
  // other code ...
  
  private def makeRequest(): Unit = {
    val now = LocalDateTime.now()
    var req: RequestQueueItem = null
    var foundValidRequest = false
    do {
      req = requestQueue.pollFirst()
      if (req == null) return

JohnMurray

class ApiService[T] {
  // other code ...
  
  def cleanupRequests() {
    val now = LocalDateTime.now()
    if (now.isAfter(nextCleanupTime)) {
      requestQueue.removeIf((req) => {
        val expired = req.startDeadline.exists(now.isAfter)
        if (expired) {
          req.p.complete(Failure(new TimeoutException))

JohnMurray

class ApiService[T] {
  // other code ...
  
  private var cleanupTimeFrame: FiniteDuration = timeFrame
  private var nextCleanupTime: LocalDateTime = LocalDateTime.now()
  
  def withCleanupWindow(window: FiniteDuration): ApiService[T] = {
    cleanupTimeFrame = window;
    this
  }

JohnMurray

package main

import (
	"bytes"
	"encoding/hex"
	"flag"
	"fmt"
	"io"
	"log"
	"net"

JohnMurray

import java.time.LocalDateTime
import java.util.{Deque, LinkedList}
import scala.concurrent.{Future, Promise}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}

import java.util.concurrent.{ScheduledThreadPoolExecutor, ScheduledFuture, TimeUnit}
import java.time.temporal.ChronoUnit

JohnMurray

import java.util.concurrent.{ScheduledThreadPoolExecutor, ScheduledFuture, TimeUnit}
import java.time.temporal.ChronoUnit

class ApiService[T](limit: Long, timeFrame: FiniteDuration) {
  // previous code ...
  
  private val timerPool = new ScheduledThreadPoolExecutor(1)
  @volatile private var recheckFut: Option[ScheduledFuture[_]] = None
  
  private def tryRequest(): Unit = this.synchronized {

JohnMurray

import java.time.LocalDateTime
import java.util.{Deque, LinkedList}
import scala.concurrent.{Future, Promise}
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import scala.util.{Failure, Success}

import java.util.concurrent.{ScheduledThreadPoolExecutor, ScheduledFuture, TimeUnit}
import java.time.temporal.ChronoUnit

JohnMurray

class ApiService[T](limit: Long, timeFrame: FiniteDuration) {
  // other code ...

  private def makeRequest(): Unit = {
    val req = requestQueue.pollFirst()
    if (req == null) {
      return
    }

    requestCount += 1