// Path: src/main/java/io/advantageous/reakt/exception/RejectedStreamException.java
// public class RejectedStreamException extends RuntimeException {
// 
//     public RejectedStreamException() {
//         super();
//     }
// 
//     public RejectedStreamException(String message) {
//         super(message);
//     }
// 
//     public RejectedStreamException(String message, Throwable cause) {
//         super(message, cause);
//     }
// 
//     public RejectedStreamException(Throwable cause) {
//         super(cause);
//     }
// 
//     protected RejectedStreamException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
//         super(message, cause, enableSuppression, writableStackTrace);
//     }
// }
// 
// Path: src/main/java/io/advantageous/reakt/impl/StreamResultImpl.java
// public class StreamResultImpl<T> extends ResultImpl<T> implements StreamResult<T> {
//     private final boolean done;
//     private final Expected<Runnable> cancelCallback;
//     private final Expected<Consumer<Long>> requestMore;
// 
//     public StreamResultImpl(final Object object,
//                             final boolean done,
//                             final Expected<Runnable> cancelCallback,
//                             final Expected<Consumer<Long>> requestMore) {
//         super(object);
//         this.done = done;
//         this.cancelCallback = cancelCallback;
//         this.requestMore = requestMore;
//     }
// 
//     @Override
//     public boolean complete() {
//         return done;
//     }
// 
//     @Override
//     public void cancel() {
//         cancelCallback.ifPresent(Runnable::run);
//     }
// 
//     @Override
//     public void request(long n) {
//         requestMore.ifPresent(longConsumer -> longConsumer.accept(n));
//     }
// }


import io.advantageous.reakt.exception.RejectedStreamException;
import io.advantageous.reakt.impl.StreamResultImpl;
import java.util.function.Consumer;
/*
 *
 *  Copyright (c) 2016. Rick Hightower, Geoff Chandler
 *
 *  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
 *
 *    		http://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 io.advantageous.reakt;



/**
 * A generic event handler for N results, i.e., a stream of results.
 *
 * This is a like a type of {@link CallbackHandler} for streaming results.
 * While {@code CallbackHandler} can be considered for scalar results, a
 * {@code Stream} is more appropriate for non-scalar results, i.e., {@code Stream#onNext}
 * will get called many times which can be thought of as a callback handler.
 * This is like an async future or promise.
 *
 * @param <T> type of result returned from callback
 * @author Rick Hightower
 */
public interface Stream<T> extends Callback<T> {

    /**
     * (Client view)
     * A result was returned so handle it.
     * This is registered from the callers (or event receivers perspective).
     * A client of a service would override {@code onResult}.
     *
     * @param result to handle
     */
    void onNext(final StreamResult<T> result);

    /**
     * (Service view)
     * This allows services to send back a last result easily to the client/handler.
     * This is a helper methods for producers (services that produce results) to send a result.
     *
     * @param result result value to send.
     */
    default void complete(final T result) {
// Path: src/main/java/io/advantageous/reakt/exception/RejectedStreamException.java
// public class RejectedStreamException extends RuntimeException {
// 
//     public RejectedStreamException() {
//         super();
//     }
// 
//     public RejectedStreamException(String message) {
//         super(message);
//     }
// 
//     public RejectedStreamException(String message, Throwable cause) {
//         super(message, cause);
//     }
// 
//     public RejectedStreamException(Throwable cause) {
//         super(cause);
//     }
// 
//     protected RejectedStreamException(String message, Throwable cause, boolean enableSuppression, boolean writableStackTrace) {
//         super(message, cause, enableSuppression, writableStackTrace);
//     }
// }
// 
// Path: src/main/java/io/advantageous/reakt/impl/StreamResultImpl.java
// public class StreamResultImpl<T> extends ResultImpl<T> implements StreamResult<T> {
//     private final boolean done;
//     private final Expected<Runnable> cancelCallback;
//     private final Expected<Consumer<Long>> requestMore;
// 
//     public StreamResultImpl(final Object object,
//                             final boolean done,
//                             final Expected<Runnable> cancelCallback,
//                             final Expected<Consumer<Long>> requestMore) {
//         super(object);
//         this.done = done;
//         this.cancelCallback = cancelCallback;
//         this.requestMore = requestMore;
//     }
// 
//     @Override
//     public boolean complete() {
//         return done;
//     }
// 
//     @Override
//     public void cancel() {
//         cancelCallback.ifPresent(Runnable::run);
//     }
// 
//     @Override
//     public void request(long n) {
//         requestMore.ifPresent(longConsumer -> longConsumer.accept(n));
//     }
// }


// Path: src/main/java/io/advantageous/reakt/Stream.java
import io.advantageous.reakt.exception.RejectedStreamException;
import io.advantageous.reakt.impl.StreamResultImpl;
import java.util.function.Consumer;

/*
 *
 *  Copyright (c) 2016. Rick Hightower, Geoff Chandler
 *
 *  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
 *
 *    		http://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 io.advantageous.reakt;



/**
 * A generic event handler for N results, i.e., a stream of results.
 *
 * This is a like a type of {@link CallbackHandler} for streaming results.
 * While {@code CallbackHandler} can be considered for scalar results, a
 * {@code Stream} is more appropriate for non-scalar results, i.e., {@code Stream#onNext}
 * will get called many times which can be thought of as a callback handler.
 * This is like an async future or promise.
 *
 * @param <T> type of result returned from callback
 * @author Rick Hightower
 */
public interface Stream<T> extends Callback<T> {

    /**
     * (Client view)
     * A result was returned so handle it.
     * This is registered from the callers (or event receivers perspective).
     * A client of a service would override {@code onResult}.
     *
     * @param result to handle
     */
    void onNext(final StreamResult<T> result);

    /**
     * (Service view)
     * This allows services to send back a last result easily to the client/handler.
     * This is a helper methods for producers (services that produce results) to send a result.
     *
     * @param result result value to send.
     */
    default void complete(final T result) {
        this.onNext(new StreamResultImpl<>(result, true, Expected.empty(), Expected.empty()));