exercises

/**
 * Async Iterators & Streams - Exercises
 * Practice working with async data sources
 */

const delay = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// =============================================================================
// EXERCISE 1: Async Range Generator
// Create an async generator that yields numbers in a range
// =============================================================================

/*
 * TODO: Create asyncRange(start, end, options) that:
 * - Yields numbers from start to end (inclusive)
 * - options.step: increment amount (default 1)
 * - options.delay: ms between yields (default 0)
 * - Supports negative ranges (counting down)
 */

async function* asyncRange(start, end, options = {}) {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
async function* asyncRange(start, end, options = {}) {
    const { step = 1, delay: delayMs = 0 } = options;
    
    if (start <= end) {
        for (let i = start; i <= end; i += Math.abs(step)) {
            if (delayMs > 0) await delay(delayMs);
            yield i;
        }
    } else {
        for (let i = start; i >= end; i -= Math.abs(step)) {
            if (delayMs > 0) await delay(delayMs);
            yield i;
        }
    }
}
*/

// =============================================================================
// EXERCISE 2: Async Iterator Pipeline
// Build a fluent pipeline for async iterators
// =============================================================================

/*
 * TODO: Create AsyncPipeline class that:
 * - Wraps an async iterable
 * - Provides chainable methods: map, filter, take, skip
 * - Has terminal methods: toArray, reduce, forEach
 * - Is lazy (only processes on terminal call)
 */

class AsyncPipeline {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
class AsyncPipeline {
    constructor(source) {
        this.source = source;
        this.transforms = [];
    }
    
    static from(source) {
        return new AsyncPipeline(source);
    }
    
    map(fn) {
        this.transforms.push({
            type: 'map',
            fn
        });
        return this;
    }
    
    filter(predicate) {
        this.transforms.push({
            type: 'filter',
            fn: predicate
        });
        return this;
    }
    
    take(n) {
        this.transforms.push({
            type: 'take',
            n
        });
        return this;
    }
    
    skip(n) {
        this.transforms.push({
            type: 'skip',
            n
        });
        return this;
    }
    
    async *[Symbol.asyncIterator]() {
        let iterator = this.source[Symbol.asyncIterator]
            ? this.source[Symbol.asyncIterator]()
            : this.source;
        
        let count = 0;
        let skipped = 0;
        let taken = 0;
        
        const skipAmount = this.transforms
            .filter(t => t.type === 'skip')
            .reduce((sum, t) => sum + t.n, 0);
        
        const takeAmount = this.transforms
            .filter(t => t.type === 'take')
            .map(t => t.n)[0] || Infinity;
        
        const mapFns = this.transforms
            .filter(t => t.type === 'map')
            .map(t => t.fn);
        
        const filterFns = this.transforms
            .filter(t => t.type === 'filter')
            .map(t => t.fn);
        
        for await (let item of this.source) {
            // Skip
            if (skipped < skipAmount) {
                skipped++;
                continue;
            }
            
            // Take limit
            if (taken >= takeAmount) {
                break;
            }
            
            // Apply maps
            for (const mapFn of mapFns) {
                item = await mapFn(item);
            }
            
            // Apply filters
            let passes = true;
            for (const filterFn of filterFns) {
                if (!(await filterFn(item))) {
                    passes = false;
                    break;
                }
            }
            
            if (passes) {
                yield item;
                taken++;
            }
        }
    }
    
    async toArray() {
        const result = [];
        for await (const item of this) {
            result.push(item);
        }
        return result;
    }
    
    async reduce(reducer, initial) {
        let accumulator = initial;
        for await (const item of this) {
            accumulator = await reducer(accumulator, item);
        }
        return accumulator;
    }
    
    async forEach(fn) {
        for await (const item of this) {
            await fn(item);
        }
    }
    
    async first() {
        for await (const item of this) {
            return item;
        }
        return undefined;
    }
    
    async count() {
        let count = 0;
        for await (const item of this) {
            count++;
        }
        return count;
    }
}
*/

// =============================================================================
// EXERCISE 3: Buffered Async Iterator
// Create an iterator that buffers ahead
// =============================================================================

/*
 * TODO: Create BufferedIterator class that:
 * - Prefetches n items ahead
 * - Reduces wait time for consumers
 * - Properly handles backpressure
 * - Cleans up on early termination
 */

class BufferedIterator {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
class BufferedIterator {
    constructor(source, bufferSize = 3) {
        this.source = source[Symbol.asyncIterator]();
        this.bufferSize = bufferSize;
        this.buffer = [];
        this.filling = false;
        this.done = false;
        this.error = null;
    }
    
    async fillBuffer() {
        if (this.filling || this.done) return;
        this.filling = true;
        
        try {
            while (this.buffer.length < this.bufferSize && !this.done) {
                const result = await this.source.next();
                
                if (result.done) {
                    this.done = true;
                    break;
                }
                
                this.buffer.push(result.value);
            }
        } catch (error) {
            this.error = error;
        } finally {
            this.filling = false;
        }
    }
    
    async next() {
        // Start filling if needed
        if (this.buffer.length < this.bufferSize && !this.done) {
            this.fillBuffer(); // Don't await - let it fill in background
        }
        
        // Wait for at least one item
        while (this.buffer.length === 0 && !this.done && !this.error) {
            await delay(1);
        }
        
        if (this.error) {
            throw this.error;
        }
        
        if (this.buffer.length > 0) {
            const value = this.buffer.shift();
            
            // Trigger refill
            if (this.buffer.length < this.bufferSize && !this.done) {
                this.fillBuffer();
            }
            
            return { value, done: false };
        }
        
        return { value: undefined, done: true };
    }
    
    async return() {
        this.done = true;
        this.buffer = [];
        if (this.source.return) {
            await this.source.return();
        }
        return { value: undefined, done: true };
    }
    
    [Symbol.asyncIterator]() {
        return this;
    }
}
*/

// =============================================================================
// EXERCISE 4: Async Iterator Multiplexer
// Combine multiple async sources with priority
// =============================================================================

/*
 * TODO: Create Multiplexer class that:
 * - Combines multiple async iterables
 * - Supports priority ordering
 * - Handles source completion
 * - Allows adding/removing sources dynamically
 */

class Multiplexer {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
class Multiplexer {
    constructor() {
        this.sources = new Map();
        this.pending = [];
        this.closed = false;
        this.nextId = 0;
    }
    
    addSource(iterable, priority = 0) {
        const id = this.nextId++;
        const iterator = iterable[Symbol.asyncIterator]();
        
        this.sources.set(id, {
            iterator,
            priority,
            done: false
        });
        
        // Queue initial fetch
        this.queueFetch(id);
        
        return id;
    }
    
    removeSource(id) {
        const source = this.sources.get(id);
        if (source) {
            source.done = true;
            if (source.iterator.return) {
                source.iterator.return();
            }
            this.sources.delete(id);
        }
    }
    
    queueFetch(id) {
        const source = this.sources.get(id);
        if (!source || source.done) return;
        
        source.iterator.next().then(result => {
            if (!result.done && !source.done) {
                this.pending.push({
                    id,
                    value: result.value,
                    priority: source.priority
                });
                this.queueFetch(id);
            } else {
                source.done = true;
            }
        });
    }
    
    async next() {
        while (true) {
            if (this.closed) {
                return { value: undefined, done: true };
            }
            
            // Sort by priority and get highest
            if (this.pending.length > 0) {
                this.pending.sort((a, b) => b.priority - a.priority);
                const item = this.pending.shift();
                return { value: { sourceId: item.id, value: item.value }, done: false };
            }
            
            // Check if all sources are done
            const allDone = [...this.sources.values()].every(s => s.done);
            if (allDone && this.pending.length === 0) {
                return { value: undefined, done: true };
            }
            
            // Wait for more data
            await delay(1);
        }
    }
    
    close() {
        this.closed = true;
        for (const [id] of this.sources) {
            this.removeSource(id);
        }
    }
    
    [Symbol.asyncIterator]() {
        return this;
    }
}
*/

// =============================================================================
// EXERCISE 5: Windowed Async Iterator
// Create time or count-based windows over async data
// =============================================================================

/*
 * TODO: Create WindowedIterator that:
 * - Groups items by time window or count
 * - options.type: 'count' or 'time'
 * - options.size: window size (items or ms)
 * - options.slide: sliding window support
 * - Yields arrays of items in each window
 */

class WindowedIterator {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
class WindowedIterator {
    constructor(source, options = {}) {
        this.source = source;
        this.type = options.type || 'count';
        this.size = options.size || 5;
        this.slide = options.slide || this.size;
    }
    
    async *countWindows() {
        let window = [];
        
        for await (const item of this.source) {
            window.push(item);
            
            if (window.length >= this.size) {
                yield [...window];
                window = window.slice(this.slide);
            }
        }
        
        // Yield remaining items
        if (window.length > 0) {
            yield window;
        }
    }
    
    async *timeWindows() {
        let window = [];
        let windowStart = Date.now();
        
        const iterator = this.source[Symbol.asyncIterator]();
        
        while (true) {
            const timeRemaining = this.size - (Date.now() - windowStart);
            
            if (timeRemaining <= 0) {
                if (window.length > 0) {
                    yield [...window];
                    window = window.slice(
                        Math.floor(window.length * (this.slide / this.size))
                    );
                }
                windowStart = Date.now();
            }
            
            // Try to get next item with timeout
            const result = await Promise.race([
                iterator.next(),
                delay(Math.max(timeRemaining, 1)).then(() => ({ timeout: true }))
            ]);
            
            if (result.timeout) continue;
            if (result.done) break;
            
            window.push(result.value);
        }
        
        // Yield remaining
        if (window.length > 0) {
            yield window;
        }
    }
    
    [Symbol.asyncIterator]() {
        if (this.type === 'time') {
            return this.timeWindows();
        }
        return this.countWindows();
    }
}
*/

// =============================================================================
// EXERCISE 6: Async Observable-like Stream
// Create an Observable-like async stream
// =============================================================================

/*
 * TODO: Create AsyncObservable class that:
 * - Supports multiple subscribers
 * - Has operators: map, filter, debounce, throttle
 * - Supports unsubscription
 * - Handles errors and completion
 */

class AsyncObservable {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
class AsyncObservable {
    constructor(producer) {
        this.producer = producer;
        this.operators = [];
    }
    
    static from(source) {
        return new AsyncObservable(async (subscriber) => {
            try {
                for await (const item of source) {
                    if (subscriber.closed) break;
                    subscriber.next(item);
                }
                subscriber.complete();
            } catch (error) {
                subscriber.error(error);
            }
        });
    }
    
    static interval(ms) {
        return new AsyncObservable(async (subscriber) => {
            let count = 0;
            while (!subscriber.closed) {
                await delay(ms);
                if (!subscriber.closed) {
                    subscriber.next(count++);
                }
            }
        });
    }
    
    pipe(...operators) {
        const observable = new AsyncObservable(this.producer);
        observable.operators = [...this.operators, ...operators];
        return observable;
    }
    
    subscribe(observerOrNext, error, complete) {
        const observer = typeof observerOrNext === 'function'
            ? { next: observerOrNext, error, complete }
            : observerOrNext;
        
        const subscriber = {
            closed: false,
            next: (value) => !subscriber.closed && observer.next?.(value),
            error: (err) => !subscriber.closed && observer.error?.(err),
            complete: () => {
                if (!subscriber.closed) {
                    subscriber.closed = true;
                    observer.complete?.();
                }
            }
        };
        
        // Apply operators
        let finalSubscriber = subscriber;
        for (const operator of this.operators.reverse()) {
            finalSubscriber = operator(finalSubscriber);
        }
        
        // Start producing
        this.producer(finalSubscriber);
        
        return {
            unsubscribe: () => {
                subscriber.closed = true;
            }
        };
    }
    
    // Static operators
    static map(fn) {
        return (subscriber) => ({
            ...subscriber,
            next: (value) => subscriber.next(fn(value))
        });
    }
    
    static filter(predicate) {
        return (subscriber) => ({
            ...subscriber,
            next: (value) => predicate(value) && subscriber.next(value)
        });
    }
    
    static take(count) {
        let taken = 0;
        return (subscriber) => ({
            ...subscriber,
            next: (value) => {
                if (taken < count) {
                    taken++;
                    subscriber.next(value);
                    if (taken >= count) {
                        subscriber.complete();
                    }
                }
            }
        });
    }
}
*/

// =============================================================================
// EXERCISE 7: Retry-able Async Iterator
// Create an iterator that retries on failure
// =============================================================================

/*
 * TODO: Create RetryIterator that:
 * - Wraps a potentially failing async iterator
 * - Retries failed iterations
 * - Supports exponential backoff
 * - Allows custom retry conditions
 */

class RetryIterator {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
class RetryIterator {
    constructor(source, options = {}) {
        this.sourceFactory = typeof source === 'function' ? source : () => source;
        this.maxRetries = options.maxRetries || 3;
        this.baseDelay = options.baseDelay || 100;
        this.maxDelay = options.maxDelay || 5000;
        this.shouldRetry = options.shouldRetry || (() => true);
        
        this.iterator = null;
        this.retryCount = 0;
        this.position = 0;
        this.buffer = [];
    }
    
    async initialize() {
        const source = await this.sourceFactory();
        this.iterator = source[Symbol.asyncIterator]();
    }
    
    async next() {
        if (!this.iterator) {
            await this.initialize();
        }
        
        while (true) {
            try {
                const result = await this.iterator.next();
                
                if (result.done) {
                    return result;
                }
                
                // Success - reset retry count and buffer
                this.retryCount = 0;
                this.buffer.push(result.value);
                this.position++;
                
                return result;
            } catch (error) {
                if (!this.shouldRetry(error) || this.retryCount >= this.maxRetries) {
                    throw error;
                }
                
                // Calculate delay with exponential backoff
                const delayTime = Math.min(
                    this.baseDelay * Math.pow(2, this.retryCount),
                    this.maxDelay
                );
                
                console.log(`Retry ${this.retryCount + 1}/${this.maxRetries} after ${delayTime}ms`);
                await delay(delayTime);
                
                this.retryCount++;
                
                // Reinitialize and skip already yielded items
                await this.initialize();
                for (let i = 0; i < this.position; i++) {
                    await this.iterator.next();
                }
            }
        }
    }
    
    [Symbol.asyncIterator]() {
        return this;
    }
}
*/

// =============================================================================
// EXERCISE 8: Async Queue Stream
// Create a queue-based async stream
// =============================================================================

/*
 * TODO: Create AsyncQueue that:
 * - Allows pushing items asynchronously
 * - Supports async iteration
 * - Has configurable buffer limits
 * - Handles backpressure
 */

class AsyncQueue {
  // YOUR CODE HERE:
}

// SOLUTION:
/*
class AsyncQueue {
    constructor(options = {}) {
        this.maxSize = options.maxSize || Infinity;
        this.buffer = [];
        this.waiters = [];
        this.closed = false;
        this.drainWaiters = [];
    }
    
    async push(item) {
        if (this.closed) {
            throw new Error('Queue is closed');
        }
        
        // Wait if buffer is full
        while (this.buffer.length >= this.maxSize) {
            await new Promise(resolve => this.drainWaiters.push(resolve));
        }
        
        this.buffer.push(item);
        
        // Notify any waiting consumers
        if (this.waiters.length > 0) {
            const waiter = this.waiters.shift();
            waiter.resolve({ value: this.buffer.shift(), done: false });
            
            // Signal that there's space
            if (this.drainWaiters.length > 0) {
                const drainWaiter = this.drainWaiters.shift();
                drainWaiter();
            }
        }
    }
    
    async next() {
        if (this.buffer.length > 0) {
            const value = this.buffer.shift();
            
            // Signal that there's space
            if (this.drainWaiters.length > 0) {
                const drainWaiter = this.drainWaiters.shift();
                drainWaiter();
            }
            
            return { value, done: false };
        }
        
        if (this.closed) {
            return { value: undefined, done: true };
        }
        
        // Wait for data
        return new Promise(resolve => {
            this.waiters.push({ resolve });
        });
    }
    
    close() {
        this.closed = true;
        
        // Resolve all waiting consumers
        for (const waiter of this.waiters) {
            waiter.resolve({ value: undefined, done: true });
        }
        this.waiters = [];
        
        // Resolve drain waiters
        for (const waiter of this.drainWaiters) {
            waiter();
        }
        this.drainWaiters = [];
    }
    
    get size() {
        return this.buffer.length;
    }
    
    get isClosed() {
        return this.closed;
    }
    
    [Symbol.asyncIterator]() {
        return this;
    }
}
*/

// =============================================================================
// TEST YOUR IMPLEMENTATIONS
// =============================================================================

async function runTests() {
  console.log('Testing Async Iterators & Streams...\n');

  // Test async generator
  async function* testGenerator() {
    yield 1;
    await delay(10);
    yield 2;
    await delay(10);
    yield 3;
  }

  console.log('1. Testing AsyncRange:');
  // Test code here

  console.log('\n2. Testing AsyncPipeline:');
  // Test code here

  console.log('\n3. Testing BufferedIterator:');
  // Test code here

  console.log('\nAll tests complete!');
}

// Uncomment to run tests
// runTests();