Reactive Programlama & Resilience Tasarım Patternleri

// Basit, değişmez bir veri taşıyıcısı
public record OrderRequest(String orderId, String customerId, BigDecimal amount) {}

public record OrderResult(String orderId, String status, Instant processedAt) {}

public sealed interface Result<T> permits Result.Success, Result.Failure {
    record Success<T>(T value) implements Result<T> {}
    record Failure<T>(Throwable error, String reason) implements Result<T> {}
}

public String handle(Result<OrderResult> result) {
    return switch (result) {
        case Result.Success<OrderResult> s when s.value().status().equals("PAID") ->
            "Sipariş " + s.value().orderId() + " başarıyla ödendi";

        case Result.Success<OrderResult>(var order) ->
            "Sipariş " + order.orderId() + " şu durumla işlendi: " + order.status();

        case Result.Failure<OrderResult>(var error, var reason)
            when error instanceof TimeoutException ->
            "Sipariş zaman aşımına uğradı: " + reason;

        case Result.Failure<OrderResult>(var error, var reason) ->
            "Sipariş başarısız oldu: " + reason;
    };
}

try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    List<Future<OrderResult>> futures = orders.stream()
        .map(order -> executor.submit(() -> processOrder(order)))
        .toList();

    for (Future<OrderResult> future : futures) {
        System.out.println(future.get());
    }
}

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<PaymentResult> payment = scope.fork(() -> chargePayment(order));
    Future<InventoryResult> inventory = scope.fork(() -> reserveInventory(order));

    scope.join();           // ikisini de bekle
    scope.throwIfFailed();  // ilk hatayı yukarı fırlat

    return new OrderResult(order.orderId(), "CONFIRMED", Instant.now());
}

public Mono<Result<OrderResult>> processOrderReactive(OrderRequest request) {
    return paymentClient.charge(request)
        .map(payment -> (Result<OrderResult>) new Result.Success<>(
            new OrderResult(request.orderId(), "PAID", Instant.now())))
        .onErrorResume(ex -> Mono.just(new Result.Failure<>(ex, ex.getMessage())))
        .timeout(Duration.ofSeconds(3));
}

public Result<OrderResult> processOrderBlocking(OrderRequest request) {
    try {
        var payment = paymentClient.chargeBlocking(request); // virtual thread'i bloklar - ucuzdur
        return new Result.Success<>(new OrderResult(request.orderId(), "PAID", Instant.now()));
    } catch (Exception ex) {
        return new Result.Failure<>(ex, ex.getMessage());
    }
}

public record RetryPolicy(int maxAttempts, Duration initialDelay, double backoffMultiplier) {

    public <T> Result<T> execute(Supplier<Result<T>> operation) {
        Duration delay = initialDelay;
        Result<T> last = null;

        for (int attempt = 1; attempt <= maxAttempts; attempt++) {
            last = operation.get();
            if (last instanceof Result.Success<T>) {
                return last;
            }
            if (attempt < maxAttempts) {
                try {
                    Thread.sleep(delay); // virtual thread - park etmek ucuz
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                delay = delay.multipliedBy((long) backoffMultiplier);
            }
        }
        return last;
    }
}

RetryConfig config = RetryConfig.custom()
    .maxAttempts(3)
    .waitDuration(Duration.ofMillis(500))
    .build();

Retry retry = Retry.of("orderService", config);
Supplier<OrderResult> decorated = Retry.decorateSupplier(retry, () -> processOrder(request));

public final class CircuitBreaker {
    public sealed interface State permits State.Closed, State.Open, State.HalfOpen {
        record Closed(int failureCount) implements State {}
        record Open(Instant openedAt) implements State {}
        record HalfOpen() implements State {}
    }

    private volatile State state = new State.Closed(0);
    private final int failureThreshold;
    private final Duration openDuration;

    public CircuitBreaker(int failureThreshold, Duration openDuration) {
        this.failureThreshold = failureThreshold;
        this.openDuration = openDuration;
    }

    public <T> Result<T> execute(Supplier<Result<T>> operation) {
        return switch (state) {
            case State.Open(var openedAt) when Duration.between(openedAt, Instant.now()).compareTo(openDuration) < 0 ->
                new Result.Failure<>(new IllegalStateException("Circuit open"), "circuit-open");

            case State.Open ignored -> {
                state = new State.HalfOpen();
                yield attempt(operation);
            }

            case State.HalfOpen ignored -> attempt(operation);

            case State.Closed(var failures) -> attempt(operation);
        };
    }

    private <T> Result<T> attempt(Supplier<Result<T>> operation) {
        Result<T> result = operation.get();
        state = switch (result) {
            case Result.Success<T> s -> new State.Closed(0);
            case Result.Failure<T> f -> {
                int failures = (state instanceof State.Closed(var count)) ? count + 1 : failureThreshold;
                yield failures >= failureThreshold
                    ? new State.Open(Instant.now())
                    : new State.Closed(failures);
            }
        };
        return result;
    }
}

CircuitBreakerConfig config = CircuitBreakerConfig.custom()
    .failureRateThreshold(50)
    .waitDurationInOpenState(Duration.ofSeconds(10))
    .slidingWindowSize(10)
    .build();

CircuitBreaker cb = CircuitBreaker.of("paymentService", config);
Supplier<OrderResult> decorated = CircuitBreaker.decorateSupplier(cb, () -> processOrder(request));

public <T> Result<T> withTimeout(Duration timeout, Callable<T> operation) {
    try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
        Future<T> future = executor.submit(operation);
        try {
            return new Result.Success<>(future.get(timeout.toMillis(), TimeUnit.MILLISECONDS));
        } catch (TimeoutException e) {
            future.cancel(true);
            return new Result.Failure<>(e, "operation-timed-out");
        } catch (Exception e) {
            return new Result.Failure<>(e, e.getMessage());
        }
    }
}

public record Bulkhead(Semaphore semaphore) {

    public static Bulkhead of(int maxConcurrentCalls) {
        return new Bulkhead(new Semaphore(maxConcurrentCalls));
    }

    public <T> Result<T> execute(Supplier<Result<T>> operation) {
        if (!semaphore.tryAcquire()) {
            return new Result.Failure<>(new RejectedExecutionException(), "bulkhead-full");
        }
        try {
            return operation.get();
        } finally {
            semaphore.release();
        }
    }
}

public final class RateLimiter {
    private final int permitsPerWindow;
    private final Duration window;
    private final AtomicInteger count = new AtomicInteger(0);
    private volatile Instant windowStart = Instant.now();

    public RateLimiter(int permitsPerWindow, Duration window) {
        this.permitsPerWindow = permitsPerWindow;
        this.window = window;
    }

    public synchronized boolean tryAcquire() {
        if (Duration.between(windowStart, Instant.now()).compareTo(window) > 0) {
            windowStart = Instant.now();
            count.set(0);
        }
        return count.incrementAndGet() <= permitsPerWindow;
    }
}

public OrderResult withFallback(Result<OrderResult> result, OrderRequest request) {
    return switch (result) {
        case Result.Success<OrderResult>(var order) -> order;
        case Result.Failure<OrderResult> f ->
            new OrderResult(request.orderId(), "PENDING_MANUAL_REVIEW", Instant.now());
    };
}

public final class ResilientOrderService {

    private final RetryPolicy retryPolicy =
        new RetryPolicy(3, Duration.ofMillis(200), 2.0);
    private final CircuitBreaker circuitBreaker =
        new CircuitBreaker(5, Duration.ofSeconds(30));
    private final Bulkhead bulkhead =
        Bulkhead.of(50);
    private final PaymentClient paymentClient;

    public ResilientOrderService(PaymentClient paymentClient) {
        this.paymentClient = paymentClient;
    }

    public OrderResult processOrder(OrderRequest request) {
        Result<OrderResult> result = bulkhead.execute(() ->
            circuitBreaker.execute(() ->
                retryPolicy.execute(() -> callPaymentWithTimeout(request))
            )
        );

        return switch (result) {
            case Result.Success<OrderResult>(var order) -> order;
            case Result.Failure<OrderResult>(var error, var reason)
                when error instanceof TimeoutException ->
                new OrderResult(request.orderId(), "TIMED_OUT_RETRY_LATER", Instant.now());
            case Result.Failure<OrderResult> f ->
                new OrderResult(request.orderId(), "FAILED: " + f.reason(), Instant.now());
        };
    }

    private Result<OrderResult> callPaymentWithTimeout(OrderRequest request) {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            Future<OrderResult> future = executor.submit(() -> {
                paymentClient.chargeBlocking(request);
                return new OrderResult(request.orderId(), "PAID", Instant.now());
            });
            return new Result.Success<>(future.get(3, TimeUnit.SECONDS));
        } catch (TimeoutException e) {
            return new Result.Failure<>(e, "payment-timeout");
        } catch (Exception e) {
            return new Result.Failure<>(e, e.getMessage());
        }
    }

    public List<OrderResult> processOrdersConcurrently(List<OrderRequest> requests) {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            List<Future<OrderResult>> futures = requests.stream()
                .map(req -> executor.submit(() -> processOrder(req)))
                .toList();

            return futures.stream()
                .map(f -> {
                    try {
                        return f.get();
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                })
                .toList();
        }
    }
}

// Thread / Runnable - platform vs virtual (JDK 21+)
Thread.ofPlatform().start(() -> System.out.println("platform"));
Thread.ofVirtual().start(() -> System.out.println("virtual"));   // ucuz, I/O dostu

// synchronized vs ReentrantLock
synchronized (lock) { counter++; }
lock.lock();
try { counter++; } finally { lock.unlock(); }

// ReadWriteLock - çok okuyucu, tek yazar
var rw = new ReentrantReadWriteLock();
rw.readLock().lock();    try { var v = cache; } finally { rw.readLock().unlock(); }
rw.writeLock().lock();   try { cache = v; }     finally { rw.writeLock().unlock(); }

// Semaphore - eşzamanlı erişimi sınırla
var sem = new Semaphore(5);
if (sem.tryAcquire()) { try { work(); } finally { sem.release(); } }

// CountDownLatch - N görevin bitmesini bekle
var latch = new CountDownLatch(3);
jobs.forEach(j -> executor.submit(() -> { j.run(); latch.countDown(); }));
latch.await();

// CyclicBarrier - thread'ler ortak bir noktada buluşur
var barrier = new CyclicBarrier(4, () -> System.out.println("hepsi geldi"));
parties.forEach(p -> executor.submit(() -> { phase1(); barrier.await(); phase2(); }));

// Atomic sınıflar (CAS) - lock-free sayaç/durum
var count = new AtomicInteger();
count.incrementAndGet();
var ref = new AtomicReference<String>("init");
ref.compareAndSet("init", "updated");

// ThreadLocal
var tl = ThreadLocal.withInitial(() -> "ctx");
String ctx = tl.get();

// ScopedValue (preview) - virtual thread dostu, yapısal paylaşım
ScopedValue.where(USER_ID, "u-1").run(() -> handleRequest());

// Fixed Thread Pool
try (var pool = Executors.newFixedThreadPool(8)) { pool.submit(task); }

// Cached Thread Pool
try (var pool = Executors.newCachedThreadPool()) { pool.submit(task); }

// Scheduled Executor - gecikmeli / periyodik
try (var sched = Executors.newScheduledThreadPool(2)) {
    sched.schedule(task, 5, TimeUnit.SECONDS);
    sched.scheduleAtFixedRate(task, 0, 1, TimeUnit.MINUTES);
}

// Work-Stealing Pool
try (var fjp = Executors.newWorkStealingPool()) { fjp.submit(task); }

// Fork/Join - böl ve fethet
class SumTask extends RecursiveTask<Long> {
    protected Long compute() {
        if (range.size() <= THRESHOLD) return range.sum();
        var left  = new SumTask(range.left()).fork();       // asenkron çalıştır
        long right = new SumTask(range.right()).compute();  // inline çalıştır
        return left.join() + right;
    }
}

// Thread Pool Per Resource - downstream bağımlılıkları izole et (bulkhead)
var paymentPool   = Executors.newFixedThreadPool(4);
var inventoryPool = Executors.newFixedThreadPool(4);

// Virtual Thread Per Task Executor - milyonlarca ucuz görev
try (var vt = Executors.newVirtualThreadPerTaskExecutor()) {
    orders.forEach(o -> vt.submit(() -> process(o)));
}

// ShutdownOnFailure - ya hep ya hiç
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    var a = scope.fork(() -> fetchA());
    var b = scope.fork(() -> fetchB());
    scope.join();            // ikisini de bekle
    scope.throwIfFailed();   // ilk hatayı yukarı fırlat
    return new Result(a.get(), b.get());
}

// ShutdownOnSuccess - ilk başarı kazanır, diğerleri iptal edilir
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
    scope.fork(() -> fetchFromCache());
    scope.fork(() -> fetchFromDb());
    scope.join();
    return scope.result();   // ilk başarılı sonuç
}

// Fan-out / Fan-in - paralel map, sonra topla
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    List<Supplier<Item>> tasks = ids.stream()
        .map(id -> scope.fork(() -> load(id)))
        .map(StructuredTaskScope.Subtask::get)
        .toList();
    scope.join();
    return tasks.stream().map(Supplier::get).toList();   // fan-in
}

// Scatter-Gather - paralel sorgu, ilk/en iyi cevabı al
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<Quote>()) {
    providers.forEach(p -> scope.fork(p::quote));
    scope.join();
    return scope.result();
}

// Future / Promise
Future<Order> f = executor.submit(() -> loadOrder(id));
Order order = f.get();

// CompletableFuture zincirleme
CompletableFuture<Order> cf = CompletableFuture
    .supplyAsync(() -> loadOrder(id))
    .thenApply(Order::total)                     // map
    .thenCompose(this::charge)                   // flatMap
    .thenCombine(loadShipping(), Receipt::new);  // iki future'ı birleştir

// Callback - tamamlanınca tepki ver
cf.whenComplete((result, err) ->
    System.out.println(err == null ? result : err.getMessage()));

// Parallel Streams
List<Result> results = orders.parallelStream()
    .map(this::process)
    .toList();

// Reactive Streams (Project Reactor)
Mono<Order> order = client.getOrder(id);
Flux<OrderEvent> events = order
    .flatMapMany(o -> eventClient.stream(o.id()))
    .onBackpressureBuffer(100);

// Event Loop - tek bir event-loop thread'i non-blocking I/O sürer
var group = new NioEventLoopGroup(1);   // tek event-loop thread
group.register(channel);

// Publish-Subscribe
publisher.subscribe(subscriber);        // tek yayıncı, çok abone

// Backpressure stratejileri
flux.onBackpressureBuffer(100);         // tamponla
flux.onBackpressureDrop();              // dolunca yeniyi bırak
flux.onBackpressureLatest();            // sadece sonuncuyu tut
flux.onBackpressureError();             // hemen hata ver

// Pipeline Pattern - art arda dönüşüm aşamaları
stream.map(Parser::parse)
      .map(Validator::validate)
      .map(Enricher::enrich)
      .forEach(sink);

// Producer-Consumer (BlockingQueue)
var queue = new LinkedBlockingQueue<Item>(100);
queue.put(item);          // üretici dolunca bekler
Item item = queue.take(); // tüketici boşken bekler

// Worker Pool / Task Queue
try (var pool = Executors.newFixedThreadPool(4)) {
    jobs.forEach(pool::submit);
}

// Guarded Suspension - koşul sağlanana kadar bekle
synchronized (lock) {
    while (!ready) lock.wait();
    use();
}

// Balking - uygun durumda değilse hemen reddet
synchronized (lock) {
    if (busy) return;     // balk
    busy = true;
}

// Two-Phase Termination
t.interrupt();                                   // faz 1: kapanma isteği
while (!Thread.currentThread().isInterrupted())  // faz 2: gözlemle ve temizle
    work();

// Double-Checked Locking
class Singleton {
    private volatile Singleton instance;
    Singleton get() {
        if (instance == null) {
            synchronized (this) {
                if (instance == null) instance = new Singleton();
            }
        }
        return instance;
    }
}

// Saga - yerel adımlar + hata durumunda telafi (compensation)
try {
    reserveInventory(); chargePayment(); createShipment();
} catch (PaymentException e) {
    releaseInventory();   // telafi et
}

// Idempotency Key - kayıtlı sonucu döndür, yeniden çalıştırma
var existing = repo.find(key);
if (existing != null) return existing;
var result = execute();
repo.save(key, result);

// Optimistic Locking - versiyon kontrolü; 0 satır güncellendiyse çakışma, retry
// UPDATE orders SET total=?, version=version+1 WHERE id=? AND version=?

// Distributed Lock (Redis) - instance'lar arası karşılıklı dışlama
boolean locked = redis.setIfAbsent("lock:order-1", token, 30, SECONDS);
if (locked) { try { work(); } finally { redis.release("lock:order-1", token); } }