Implement Circuit Breaker

type CircuitBreaker struct {
    mu              sync.Mutex
    failures        int
    threshold       int
    state           string // "closed", "open", "half-open"
    lastFailureTime time.Time
    resetTimeout    time.Duration
}

func NewCircuitBreaker(threshold int, resetTimeout time.Duration) *CircuitBreaker {
    return &CircuitBreaker{
        threshold:    threshold,
        state:        "closed",
        resetTimeout: resetTimeout,
    }
}

import time
import threading

class CircuitBreaker:
    def __init__(self, threshold=5, reset_timeout=30):
        self.threshold = threshold
        self.reset_timeout = reset_timeout
        self.failures = 0
        self.state = "closed"
        self.last_failure_time = 0
        self._lock = threading.Lock()

class CircuitBreaker {
  constructor(threshold = 5, resetTimeout = 30000) {
    this.threshold = threshold;
    this.resetTimeout = resetTimeout;
    this.failures = 0;
    this.state = "closed";
    this.lastFailureTime = 0;
  }
}

public class CircuitBreaker {
    private final int threshold;
    private final long resetTimeoutMs;
    private int failures = 0;
    private String state = "closed";
    private long lastFailureTime = 0;

    public CircuitBreaker(int threshold, long resetTimeoutMs) {
        this.threshold = threshold;
        this.resetTimeoutMs = resetTimeoutMs;
    }
}

public class CircuitBreaker
{
    private readonly int _threshold;
    private readonly TimeSpan _resetTimeout;
    private int _failures;
    private string _state = "closed";
    private DateTime _lastFailureTime;
    private readonly object _lock = new();

    public CircuitBreaker(int threshold = 5, TimeSpan? resetTimeout = null)
    {
        _threshold = threshold;
        _resetTimeout = resetTimeout ?? TimeSpan.FromSeconds(30);
    }
}

class CircuitBreaker(
    private val threshold: Int = 5,
    private val resetTimeout: Long = 30000
) {
    private var failures = 0
    private var state = "closed"
    private var lastFailureTime = 0L
    private val lock = Any()
}

class CircuitBreaker {
    int threshold;
    int resetTimeout;
    int failures = 0;
    std::string state = "closed";
    std::chrono::steady_clock::time_point lastFailureTime;
    std::mutex mtx;

public:
    CircuitBreaker(int threshold = 5, int resetTimeoutSec = 30)
        : threshold(threshold), resetTimeout(resetTimeoutSec) {}
};

use std::time::{Duration, Instant};

#[derive(Clone, Copy, PartialEq)]
enum State {
    Closed,
    Open,
    HalfOpen,
}

struct CircuitBreaker {
    threshold: u32,
    reset_timeout: Duration,
    failures: u32,
    state: State,
    last_failure: Option<Instant>,
}

impl CircuitBreaker {
    fn new(threshold: u32, reset_timeout: Duration) -> Self {
        Self {
            threshold,
            reset_timeout,
            failures: 0,
            state: State::Closed,
            last_failure: None,
        }
    }
}

class CircuitBreaker
  def initialize(threshold: 5, reset_timeout: 30)
    @threshold = threshold
    @reset_timeout = reset_timeout
    @failures = 0
    @state = :closed
    @last_failure_time = nil
    @mutex = Mutex.new
  end
end

defmodule CircuitBreaker do
  # Backed by an Agent so the state survives across calls.
  defstruct threshold: 5,
            reset_timeout_ms: 30_000,
            failures: 0,
            state: :closed,
            last_failure_ms: nil

  def start_link(threshold \\ 5, reset_timeout_ms \\ 30_000) do
    Agent.start_link(fn ->
      %CircuitBreaker{threshold: threshold, reset_timeout_ms: reset_timeout_ms}
    end)
  end
end

func (cb *CircuitBreaker) RecordFailure() {
    cb.mu.Lock()
    defer cb.mu.Unlock()
    cb.failures++
    cb.lastFailureTime = time.Now()
    if cb.failures >= cb.threshold {
        cb.state = "open"
        log.Printf("Circuit OPEN after %d failures", cb.failures)
    }
}

func (cb *CircuitBreaker) RecordSuccess() {
    cb.mu.Lock()
    defer cb.mu.Unlock()
    cb.failures = 0
    cb.state = "closed"
}

def record_failure(self):
    with self._lock:
        self.failures += 1
        self.last_failure_time = time.time()
        if self.failures >= self.threshold:
            self.state = "open"
            print(f"Circuit OPEN after {self.failures} failures")

def record_success(self):
    with self._lock:
        self.failures = 0
        self.state = "closed"

recordFailure() {
  this.failures++;
  this.lastFailureTime = Date.now();
  if (this.failures >= this.threshold) {
    this.state = "open";
    console.log(`Circuit OPEN after ${this.failures} failures`);
  }
}

recordSuccess() {
  this.failures = 0;
  this.state = "closed";
}

public synchronized void recordFailure() {
    failures++;
    lastFailureTime = System.currentTimeMillis();
    if (failures >= threshold) {
        state = "open";
        System.out.printf("Circuit OPEN after %d failures%n", failures);
    }
}

public synchronized void recordSuccess() {
    failures = 0;
    state = "closed";
}

public void RecordFailure()
{
    lock (_lock)
    {
        _failures++;
        _lastFailureTime = DateTime.UtcNow;
        if (_failures >= _threshold)
        {
            _state = "open";
            Console.WriteLine($"Circuit OPEN after {_failures} failures");
        }
    }
}

public void RecordSuccess()
{
    lock (_lock) { _failures = 0; _state = "closed"; }
}

fun recordFailure() = synchronized(lock) {
    failures++
    lastFailureTime = System.currentTimeMillis()
    if (failures >= threshold) {
        state = "open"
        println("Circuit OPEN after $failures failures")
    }
}

fun recordSuccess() = synchronized(lock) {
    failures = 0
    state = "closed"
}

void recordFailure() {
    std::lock_guard<std::mutex> lock(mtx);
    failures++;
    lastFailureTime = std::chrono::steady_clock::now();
    if (failures >= threshold) {
        state = "open";
        std::cout << "Circuit OPEN after " << failures << " failures" << std::endl;
    }
}

void recordSuccess() {
    std::lock_guard<std::mutex> lock(mtx);
    failures = 0;
    state = "closed";
}

impl CircuitBreaker {
    fn record_failure(&mut self) {
        self.failures += 1;
        self.last_failure = Some(Instant::now());
        if self.failures >= self.threshold {
            self.state = State::Open;
            println!("Circuit OPEN after {} failures", self.failures);
        }
    }

    fn record_success(&mut self) {
        self.failures = 0;
        self.state = State::Closed;
    }
}

def record_failure
  @mutex.synchronize do
    @failures += 1
    @last_failure_time = Time.now
    if @failures >= @threshold
      @state = :open
      puts "Circuit OPEN after #{@failures} failures"
    end
  end
end

def record_success
  @mutex.synchronize do
    @failures = 0
    @state = :closed
  end
end

def record_failure(cb) do
  Agent.update(cb, fn s ->
    failures = s.failures + 1
    state = if failures >= s.threshold, do: :open, else: s.state
    if state == :open, do: IO.puts("Circuit OPEN after #{failures} failures")
    %{s | failures: failures, state: state, last_failure_ms: System.monotonic_time(:millisecond)}
  end)
end

def record_success(cb) do
  Agent.update(cb, fn s -> %{s | failures: 0, state: :closed} end)
end

func (cb *CircuitBreaker) AllowRequest() bool {
    cb.mu.Lock()
    defer cb.mu.Unlock()

    switch cb.state {
    case "closed":
        return true
    case "open":
        if time.Since(cb.lastFailureTime) > cb.resetTimeout {
            cb.state = "half-open"
            log.Println("Circuit HALF-OPEN — probing...")
            return true
        }
        return false
    case "half-open":
        return true
    }
    return false
}

def allow_request(self):
    with self._lock:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.reset_timeout:
                self.state = "half-open"
                print("Circuit HALF-OPEN — probing...")
                return True
            return False
        return True  # half-open allows one probe

allowRequest() {
  if (this.state === "closed") return true;
  if (this.state === "open") {
    if (Date.now() - this.lastFailureTime > this.resetTimeout) {
      this.state = "half-open";
      console.log("Circuit HALF-OPEN — probing...");
      return true;
    }
    return false;
  }
  return true; // half-open allows one probe
}

public synchronized boolean allowRequest() {
    if ("closed".equals(state)) return true;
    if ("open".equals(state)) {
        if (System.currentTimeMillis() - lastFailureTime > resetTimeoutMs) {
            state = "half-open";
            System.out.println("Circuit HALF-OPEN — probing...");
            return true;
        }
        return false;
    }
    return true;
}

public bool AllowRequest()
{
    lock (_lock)
    {
        if (_state == "closed") return true;
        if (_state == "open")
        {
            if (DateTime.UtcNow - _lastFailureTime > _resetTimeout)
            {
                _state = "half-open";
                Console.WriteLine("Circuit HALF-OPEN — probing...");
                return true;
            }
            return false;
        }
        return true;
    }
}

fun allowRequest(): Boolean = synchronized(lock) {
    when (state) {
        "closed" -> true
        "open" -> {
            if (System.currentTimeMillis() - lastFailureTime > resetTimeout) {
                state = "half-open"
                println("Circuit HALF-OPEN — probing...")
                true
            } else false
        }
        else -> true
    }
}

bool allowRequest() {
    std::lock_guard<std::mutex> lock(mtx);
    if (state == "closed") return true;
    if (state == "open") {
        auto elapsed = std::chrono::steady_clock::now() - lastFailureTime;
        if (elapsed > std::chrono::seconds(resetTimeout)) {
            state = "half-open";
            return true;
        }
        return false;
    }
    return true;
}

impl CircuitBreaker {
    fn allow_request(&mut self) -> bool {
        match self.state {
            State::Closed => true,
            State::Open => {
                let expired = self
                    .last_failure
                    .map(|t| t.elapsed() > self.reset_timeout)
                    .unwrap_or(true);
                if expired {
                    self.state = State::HalfOpen;
                    println!("Circuit HALF-OPEN — probing...");
                    true
                } else {
                    false
                }
            }
            State::HalfOpen => true, // allow one probe
        }
    }
}

def allow_request?
  @mutex.synchronize do
    case @state
    when :closed
      true
    when :open
      if Time.now - @last_failure_time > @reset_timeout
        @state = :half_open
        puts "Circuit HALF-OPEN — probing..."
        true
      else
        false
      end
    else
      true # half-open allows one probe
    end
  end
end

def allow_request?(cb) do
  Agent.get_and_update(cb, fn s ->
    case s.state do
      :closed ->
        {true, s}

      :open ->
        now = System.monotonic_time(:millisecond)
        if now - (s.last_failure_ms || 0) > s.reset_timeout_ms do
          IO.puts("Circuit HALF-OPEN — probing...")
          {true, %{s | state: :half_open}}
        else
          {false, s}
        end

      :half_open ->
        {true, s} # allow one probe
    end
  end)
end

cb := NewCircuitBreaker(5, 30*time.Second)

func sendCommand(ctx context.Context, client *kubemq.Client,
    body []byte) (*kubemq.CommandResponse, error) {
    if !cb.AllowRequest() {
        return nil, fmt.Errorf("circuit open — service unavailable")
    }

    resp, err := client.SendCommand(ctx, kubemq.NewCommand().
        SetChannel("orders.process").
        SetBody(body).
        SetTimeout(5 * time.Second))

    if err != nil || !resp.Executed {
        cb.RecordFailure()
        return resp, err
    }

    cb.RecordSuccess()
    return resp, nil
}

cb = CircuitBreaker(threshold=5, reset_timeout=30)

def send_command(client, body):
    if not cb.allow_request():
        raise RuntimeError("circuit open — service unavailable")

    try:
        response = client.send_command(CommandMessage(
            channel="orders.process", body=body, timeout_in_seconds=5))
        if response.is_executed:
            cb.record_success()
            return response
        cb.record_failure()
        return response
    except Exception:
        cb.record_failure()
        raise

const cb = new CircuitBreaker(5, 30000);

async function sendCommand(client, body) {
  if (!cb.allowRequest()) {
    throw new Error("circuit open — service unavailable");
  }
  try {
    const response = await client.sendCommand({
      channel: "orders.process",
      body: Buffer.from(body),
      timeoutInSeconds: 5,
    });
    if (response.isExecuted) {
      cb.recordSuccess();
    } else {
      cb.recordFailure();
    }
    return response;
  } catch (err) {
    cb.recordFailure();
    throw err;
  }
}

CircuitBreaker cb = new CircuitBreaker(5, 30000);

CommandResponseMessage sendCommand(CQClient client, byte[] body) {
    if (!cb.allowRequest()) {
        throw new RuntimeException("circuit open — service unavailable");
    }
    try {
        var resp = client.sendCommandRequest(CommandMessage.builder()
            .channel("orders.process").body(body).timeout(5000).build());
        if (resp.isExecuted()) cb.recordSuccess();
        else cb.recordFailure();
        return resp;
    } catch (Exception e) {
        cb.recordFailure();
        throw e;
    }
}

var cb = new CircuitBreaker(5, TimeSpan.FromSeconds(30));

async Task<CommandResponse> SendCommand(KubeMQClient client, byte[] body)
{
    if (!cb.AllowRequest())
        throw new InvalidOperationException("circuit open — service unavailable");

    try
    {
        var resp = await client.SendCommandAsync(new CommandMessage
        {
            Channel = "orders.process", Body = body, Timeout = TimeSpan.FromSeconds(5)
        });
        if (resp.IsExecuted) cb.RecordSuccess();
        else cb.RecordFailure();
        return resp;
    }
    catch
    {
        cb.RecordFailure();
        throw;
    }
}

val cb = CircuitBreaker(threshold = 5, resetTimeout = 30000)

fun sendCommand(client: CQClient, body: ByteArray): CommandResponse {
    if (!cb.allowRequest())
        throw RuntimeException("circuit open — service unavailable")

    return try {
        val resp = client.sendCommand(CommandMessage(
            channel = "orders.process", body = body, timeout = 5000))
        if (resp.isExecuted) cb.recordSuccess() else cb.recordFailure()
        resp
    } catch (e: Exception) {
        cb.recordFailure()
        throw e
    }
}

CircuitBreaker cb(5, 30);

kubemq::CommandResponse sendCommand(kubemq::CQClient& client,
    const std::string& body) {
    if (!cb.allowRequest()) {
        throw std::runtime_error("circuit open — service unavailable");
    }
    try {
        kubemq::CommandMessage cmd;
        cmd.channel = "orders.process";
        cmd.body = body;
        cmd.timeout = 5000;
        auto resp = client.sendCommand(cmd);
        if (resp.isExecuted) cb.recordSuccess();
        else cb.recordFailure();
        return resp;
    } catch (...) {
        cb.recordFailure();
        throw;
    }
}

use kubemq::prelude::*;
use kubemq::CommandBuilder;
use std::error::Error;
use std::time::Duration;

async fn send_command(
    client: &KubemqClient,
    cb: &mut CircuitBreaker,
    body: Vec<u8>,
) -> Result<(), Box<dyn Error>> {
    if !cb.allow_request() {
        return Err("circuit open — service unavailable".into());
    }

    let command = CommandBuilder::new()
        .channel("orders.process")
        .body(body)
        .timeout(Duration::from_secs(5))
        .build();

    match client.send_command(command).await {
        Ok(resp) if resp.executed => {
            cb.record_success();
            Ok(())
        }
        Ok(_) => {
            cb.record_failure();
            Ok(())
        }
        Err(e) => {
            cb.record_failure();
            Err(Box::new(e))
        }
    }
}

require 'kubemq'

cb = CircuitBreaker.new(threshold: 5, reset_timeout: 30)

def send_command(client, cb, body)
  raise 'circuit open — service unavailable' unless cb.allow_request?

  begin
    msg = KubeMQ::CQ::CommandMessage.new(
      channel: 'orders.process', body: body, timeout: 5000
    )
    result = client.send_command(msg)
    if result.executed
      cb.record_success
    else
      cb.record_failure
    end
    result
  rescue KubeMQ::Error
    cb.record_failure
    raise
  end
end

def send_command(client, cb, body) do
  unless CircuitBreaker.allow_request?(cb) do
    {:error, :circuit_open}
  else
    command = KubeMQ.Command.new(
      channel: "orders.process",
      body: body,
      timeout: 5_000
    )

    case KubeMQ.Client.send_command(client, command) do
      {:ok, %{executed: true} = resp} ->
        CircuitBreaker.record_success(cb)
        {:ok, resp}

      {:ok, resp} ->
        CircuitBreaker.record_failure(cb)
        {:ok, resp}

      {:error, err} ->
        CircuitBreaker.record_failure(cb)
        {:error, err}
    end
  end
end