/*
 * Copyright 2024-2026 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.cloud.ai.graph.async.internal.reactive;

import com.alibaba.cloud.ai.graph.async.AsyncGenerator;

import java.util.concurrent.Flow;

/**
 * A {@code GeneratorPublisher} is a {@link Flow.Publisher} that generates items from an
 * asynchronous generator.
 *
 * @param <T> the type of items to be published
 */
@Deprecated
public class GeneratorPublisher<T> implements Flow.Publisher<T> {

	private final AsyncGenerator<? extends T> delegate;

	/**
	 * Constructs a new <code>GeneratorPublisher</code> with the specified async
	 * generator.
	 * @param delegate The async generator to be used by this publisher.
	 */
	public GeneratorPublisher(AsyncGenerator<? extends T> delegate) {
		this.delegate = delegate;
	}

	/**
	 * Subscribes the provided {@code Flow.Subscriber} to this signal. The subscriber
	 * receives initial subscription, handles asynchronous data flow, and manages any
	 * errors or completion signals.
	 * @param subscriber The subscriber to which the signal will be delivered.
	 */
	@Override
	public void subscribe(Flow.Subscriber<? super T> subscriber) {
		subscriber.onSubscribe(new Flow.Subscription() {
			/**
			 * Requests more elements from the upstream Publisher.
			 *
			 * <p>
			 * The Publisher calls this method to indicate that it wants more items. The
			 * parameter {@code n} specifies the number of additional items requested.
			 * @param n the number of items to request, a count greater than zero
			 */
			@Override
			public void request(long n) {
			}

			/**
			 * Cancels the operation.
			 * @throws UnsupportedOperationException if the method is not yet implemented.
			 */
			@Override
			public void cancel() {
				throw new UnsupportedOperationException("cancel is not implemented yet!");
			}
		});

		delegate.forEachAsync(subscriber::onNext).thenAccept(value -> {
			subscriber.onComplete();
		}).exceptionally(ex -> {
			subscriber.onError(ex);
			return null;
		}).join();
	}

}
