如何在Java中结合SourceWithContext和SourceQueue的功能？

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import akka.Done;
import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.japi.function.Function;
import akka.stream.Attributes;
import akka.stream.Materializer;
import akka.stream.QueueOfferResult;
import akka.stream.javadsl.*;
import akka.stream.scaladsl.Source;

public class SourceWithContextAndSourceQueueExample {

    public static void main(String[] args) {
        // 创建ActorSystem和Materializer
        ActorSystem system = ActorSystem.create();
        Materializer materializer = Materializer.createMaterializer(system);

        // 创建SourceQueue的辅助函数
        Function<ActorSystem, Materializer, Queue<SourceWithContext<String, NotUsed>>> sourceQueueFactory =
                (actorSystem, mat) -> {
                    Source<String, NotUsed> source = Source.<String>queue(10, OverflowStrategy.backpressure())
                            .mapMaterializedValue(queue -> {
                                // 在队列准备好时打印消息
                                System.out.println("Queue ready");
                                return NotUsed.getInstance();
                            });

                    return source.mapMaterializedValue(queue -> {
                        // 在队列准备好时创建SourceWithContext
                        return SourceWithContext.fromSource(queue, CompletableFuture.completedFuture(NotUsed.getInstance()));
                    }).asJava();
                };

        // 创建处理队列元素的函数
        Function<SourceWithContext<String, NotUsed>, CompletionStage<Done>> processElement =
                (sourceWithContext) -> {
                    String element = sourceWithContext.first();
                    System.out.println("Processing element: " + element);
                    return CompletableFuture.completedFuture(Done.getInstance());
                };

        // 创建队列源
        SourceQueueWithComplete<SourceWithContext<String, NotUsed>> sourceQueue =
                Source.<SourceWithContext<String, NotUsed>>queue(10, OverflowStrategy.backpressure())
                        .mapMaterializedValue(queue -> {
                            // 在队列准备好时处理元素
                            CompletionStage<Done> future = sourceQueueFactory.apply(system, materializer)
                                    .flatMapConcat(source -> source.runWith(Sink.foreach(processElement), materializer))
                                    .run(queue, materializer);

                            // 在处理完成后打印消息
                            future.whenComplete((done, throwable) -> {
                                if (throwable != null) {
                                    System.out.println("Processing failed: " + throwable.getMessage());
                                } else {
                                    System.out.println("Processing completed");
                                }
                            });

                            return queue;
                        })
                        .toMat(Sink.ignore(), Keep.left())
                        .run(materializer);

        // 使用SourceQueue发送消息
        sourceQueue.offer(SourceWithContext.make("Message 1", CompletableFuture.completedFuture(NotUsed.getInstance())));
        sourceQueue.offer(SourceWithContext.make("Message 2", CompletableFuture.completedFuture(NotUsed.getInstance())));
        sourceQueue.offer(SourceWithContext.make("Message 3", CompletableFuture.completedFuture(NotUsed.getInstance())));
    }
}