import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicInteger;

public class PerformanceBestPractices {
    
    // 1. 合理使用调度器
    public Flux<String> ioIntensiveOperation(Flux<String> inputs) {
        return inputs
            .flatMap(input -> 
                Mono.fromCallable(() -> blockingIoOperation(input))
                    .subscribeOn(Schedulers.boundedElastic()) // IO操作使用弹性线程池
            );
    }
    
    private String blockingIoOperation(String input) {
        // 模拟阻塞IO操作
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return input.toUpperCase();
    }
    
    // 2. 避免在响应式链中进行阻塞操作
    public Flux<String> nonBlockingAlternative(Flux<String> inputs) {
        return inputs
            .delayElements(Duration.ofMillis(100)) // 使用延迟而不是睡眠
            .map(String::toUpperCase);
    }
    
    // 3. 合理使用缓存
    public Mono<String> getWithCache(String key) {
        return Mono.fromCallable(() -> expensiveOperation(key))
            .cache(Duration.ofMinutes(5)) // 缓存5分钟
            .subscribeOn(Schedulers.parallel());
    }
    
    private String expensiveOperation(String key) {
        // 模拟昂贵操作
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        return "Processed: " + key;
    }
    
    // 4. 背压感知处理
    public Flux<Data> processWithBackpressureAwareness(Flux<Data> dataStream) {
        AtomicInteger counter = new AtomicInteger();
        
        return dataStream
            .onBackpressureBuffer(1000, 
                data -> System.out.println("Dropping: " + data)
            )
            .doOnNext(data -> {
                if (counter.incrementAndGet() % 100 == 0) {
                    System.out.println("Processed " + counter.get() + " items");
                }
            })
            .subscribeOn(Schedulers.parallel());
    }
    
    // 5. 合理使用flatMap与concatMap
    public Flux<Result> processWithConcurrencyControl(Flux<Request> requests) {
        return requests
            .flatMap(request -> 
                processRequest(request)
                    .subscribeOn(Schedulers.parallel()),
                10 // 最大并发数
            );
    }
    
    private Mono<Result> processRequest(Request request) {
        return Mono.fromCallable(() -> {
            // 处理请求
            return new Result("Processed: " + request.getId());
        });
    }
    
    // 6. 监控与指标收集
    public Flux<Data> processWithMetrics(Flux<Data> dataStream) {
        return dataStream
            .name("dataProcessing") // 为操作命名以便监控
            .metrics() // 收集指标
            .doOnNext(data -> {
                // 业务逻辑
            })
            .doOnError(error -> {
                // 错误处理与记录
            });
    }
}

// 简单的数据模型
class Data {
    private String content;
    // getters and setters
}

class Request {
    private String id;
    // getters and setters
}

class Result {
    private String message;
    // constructor, getters
}

package cn.tcmeta.context;

import reactor.core.publisher.Mono;
import reactor.util.context.Context;

/**
 * @author: laoren
 * @date: 2025/8/27 19:46
 * @description: 在操作符之间传递 Context
 * @version: 1.0.0
 */
public class ContextAcrossOperators {
    static class UserService {
        Mono<String> getUserName(String userId) {
            return Mono.deferContextual(ctx -> {
                String requestId = ctx.getOrDefault("requestId", "unknow");
                System.out.println("requestId: " + requestId + " - userId: " + userId);
                return Mono.just("User_" + userId);
            });
        }
    }

    static class OrderService {
        Mono<String> getOrderInfo(String orderId) {
            return Mono.deferContextual(ctx -> {
                String requestId = ctx.getOrDefault("requestId", "unknow");
                String user = ctx.getOrDefault("user", "unknow");
                System.out.println("requestId: " + requestId + " - user: " + user + " - orderId: " + orderId);
                return Mono.just("Order_" + orderId + " _for_ " + user);
            });
        }
    }

    public static void main(String[] args) {
        UserService userService = new UserService();
        OrderService orderService = new OrderService();

        // 在流处理过程中传递 Context
        Mono<String> result = Mono.just("123")
                .flatMap(userService::getUserName)
                .flatMap(userName ->
                        Mono.deferContextual(ctx -> {
                            // 将用户信息存入 Context
                            return Mono.just("order456")
                                    .flatMap(orderService::getOrderInfo)
                                    .contextWrite(Context.of("user", userName));
                        })
                )
                .contextWrite(Context.of("requestId", "req-789"));

        result.subscribe(System.out::println);
        // 输出:
        // Getting user 123 with requestId: req-789
        // Getting order order456 for User_123 with requestId: req-789
        // Order_order456_for_User_123
    }
}

import reactor.core.publisher.Mono;
import reactor.util.context.Context;
import org.springframework.web.server.ServerWebExchange;

public class WebContextExample {
    
    // 模拟的 Web 处理器
    static class WebHandler {
        Mono<String> handleRequest(ServerWebExchange exchange) {
            return Mono.just("request-data")
                .flatMap(data -> processData(data))
                .contextWrite(Context.of(
                    "exchange", exchange,
                    "requestId", exchange.getRequest().getId(),
                    "user", getCurrentUser(exchange)
                ));
        }
        
        Mono<String> processData(String data) {
            return Mono.deferContextual(ctx -> {
                ServerWebExchange exchange = ctx.get("exchange");
                String requestId = ctx.get("requestId");
                String user = ctx.get("user");
                
                System.out.println("Processing data for user: " + user + ", requestId: " + requestId);
                return Mono.just("Processed: " + data);
            });
        }
        
        String getCurrentUser(ServerWebExchange exchange) {
            // 模拟从 exchange 中获取用户信息
            return "user-alice";
        }
    }
    
    // 模拟的 ServerWebExchange
    static class MockServerWebExchange {
        private MockServerRequest request = new MockServerRequest();
        
        public MockServerRequest getRequest() {
            return request;
        }
    }
    
    static class MockServerRequest {
        public String getId() {
            return "req-12345";
        }
    }
    
    public static void main(String[] args) {
        WebHandler handler = new WebHandler();
        MockServerWebExchange exchange = new MockServerWebExchange();
        
        handler.handleRequest(exchange)
            .subscribe(System.out::println);
        // 输出: 
        // Processing data for user: user-alice, requestId: req-12345
        // Processed: request-data
    }
}

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.context.Context;
import java.util.UUID;

public class TracingContextExample {
    
    // 模拟的分布式追踪器
    static class Tracer {
        void logEvent(String event, String requestId) {
            System.out.println("[" + requestId + "] " + event);
        }
        
        Mono<String> startSpan(String name, String requestId) {
            return Mono.fromCallable(() -> {
                String spanId = UUID.randomUUID().toString();
                logEvent("Started span: " + name + " (" + spanId + ")", requestId);
                return spanId;
            });
        }
        
        Mono<Void> endSpan(String spanId, String requestId) {
            return Mono.fromRunnable(() -> 
                logEvent("Ended span: " + spanId, requestId)
            );
        }
    }
    
    public static void main(String[] args) {
        Tracer tracer = new Tracer();
        String requestId = UUID.randomUUID().toString();
        
        Flux.range(1, 3)
            .flatMap(i -> 
                Mono.deferContextual(ctx -> {
                    String currentRequestId = ctx.get("requestId");
                    
                    return tracer.startSpan("process-item-" + i, currentRequestId)
                        .flatMap(spanId -> 
                            processItem(i)
                                .doOnSuccess(result -> 
                                    tracer.endSpan(spanId, currentRequestId).subscribe()
                                )
                                .doOnError(error -> 
                                    tracer.endSpan(spanId, currentRequestId).subscribe()
                                )
                        );
                })
            )
            .contextWrite(Context.of("requestId", requestId))
            .subscribe();
        
        // 输出示例:
        // [a1b2c3d4...] Started span: process-item-1 (e5f6g7h8...)
        // [a1b2c3d4...] Ended span: e5f6g7h8...
        // [a1b2c3d4...] Started span: process-item-2 (i9j0k1l2...)
        // [a1b2c3d4...] Ended span: i9j0k1l2...
        // [a1b2c3d4...] Started span: process-item-3 (m3n4o5p6...)
        // [a1b2c3d4...] Ended span: m3n4o5p6...
    }
    
    static Mono<String> processItem(int item) {
        return Mono.just("Processed item " + item)
            .delayElement(java.time.Duration.ofMillis(100));
    }
}

import reactor.core.publisher.Mono;
import reactor.util.context.Context;

public class SecurityContextExample {
    
    // 模拟的用户认证信息
    static class Authentication {
        private final String username;
        private final String[] roles;
        
        public Authentication(String username, String[] roles) {
            this.username = username;
            this.roles = roles;
        }
        
        public String getUsername() {
            return username;
        }
        
        public String[] getRoles() {
            return roles;
        }
        
        public boolean hasRole(String role) {
            for (String r : roles) {
                if (r.equals(role)) {
                    return true;
                }
            }
            return false;
        }
    }
    
    // 模拟的资源服务
    static class ResourceService {
        Mono<String> getProtectedResource(String resourceId) {
            return Mono.deferContextual(ctx -> {
                Authentication auth = ctx.get("auth");
                
                if (!auth.hasRole("ADMIN")) {
                    return Mono.error(new SecurityException("Access denied"));
                }
                
                return Mono.just("Protected resource: " + resourceId);
            });
        }
    }
    
    public static void main(String[] args) {
        ResourceService service = new ResourceService();
        
        // 有权限的用户
        Authentication adminAuth = new Authentication("alice", new String[]{"USER", "ADMIN"});
        
        service.getProtectedResource("secret-data")
            .contextWrite(Context.of("auth", adminAuth))
            .subscribe(
                System.out::println,
                error -> System.err.println("Error: " + error.getMessage())
            );
        // 输出: Protected resource: secret-data
        
        // 无权限的用户
        Authentication userAuth = new Authentication("bob", new String[]{"USER"});
        
        service.getProtectedResource("secret-data")
            .contextWrite(Context.of("auth", userAuth))
            .subscribe(
                System.out::println,
                error -> System.err.println("Error: " + error.getMessage())
            );
        // 输出: Error: Access denied
    }
}