cell/examples/http_download_actor.ce

// HTTP Download Actor
// Handles download requests and progress queries
var http = use('http');
var os = use('os');

// Actor state
var state = {
    downloading: false,
    current_url: null,
    total_bytes: 0,
    downloaded_bytes: 0,
    start_time: 0,
    error: null,
    connection: null,
    download_msg: null,
    chunks: []
};

// Helper to calculate progress percentage
function get_progress() {
    if (state.total_bytes == 0) {
        return 0;
    }
    return Math.round((state.downloaded_bytes / state.total_bytes) * 100);
}

// Helper to format status response
function get_status() {
    if (!state.downloading) {
        return {
            status: 'idle',
            error: state.error
        };
    }
    
    var elapsed = os.now() - state.start_time;
    var bytes_per_sec = elapsed > 0 ? state.downloaded_bytes / elapsed : 0;
    
    return {
        status: 'downloading',
        url: state.current_url,
        progress: get_progress(),
        downloaded_bytes: state.downloaded_bytes,
        total_bytes: state.total_bytes,
        elapsed_seconds: elapsed,
        bytes_per_second: Math.round(bytes_per_sec)
    };
}

// Main message receiver
$_.receiver(function(msg) {
    switch (msg.type) {
        case 'download':
            if (state.downloading) {
                send(msg, {
                    type: 'error',
                    error: 'Already downloading',
                    current_url: state.current_url
                });
                return;
            }
            
            if (!msg.url) {
                send(msg, {
                    type: 'error',
                    error: 'No URL provided'
                });
                return;
            }
            
            // Start download
            state.downloading = true;
            state.current_url = msg.url;
            state.total_bytes = 0;
            state.downloaded_bytes = 0;
            state.start_time = os.now();
            state.error = null;
            state.download_msg = msg;
            state.chunks = [];
            
            try {
                // Start the connection
                state.connection = http.fetch_start(msg.url, msg.options || {});
                if (!state.connection) {
                    throw new Error('Failed to start download');
                }
                
                // Schedule the first chunk read
                $_.delay(read_next_chunk, 0);
                
            } catch (e) {
                state.error = e.toString();
                state.downloading = false;
                
                send(msg, {
                    type: 'error',
                    error: state.error,
                    url: msg.url
                });
            }
            break;
            
        case 'status':
            log.console(`got status request. current is ${json.encode(get_status())}`)
            send(msg, {
                type: 'status_response',
                ...get_status()
            });
            break;
            
        case 'cancel':
            if (state.downloading) {
                // Cancel the download
                if (state.connection) {
                    http.fetch_close(state.connection);
                    state.connection = null;
                }
                state.downloading = false;
                state.current_url = null;
                state.download_msg = null;
                state.chunks = [];
                
                send(msg, {
                    type: 'cancelled',
                    message: 'Download cancelled',
                    url: state.current_url
                });
            } else {
                send(msg, {
                    type: 'error',
                    error: 'No download in progress'
                });
            }
            break;
            
        default:
            send(msg, {
                type: 'error',
                error: 'Unknown message type: ' + msg.type
            });
    }
});

// Non-blocking chunk reader
function read_next_chunk() {
    if (!state.downloading || !state.connection) {
        return;
    }
    
    try {
        var chunk = http.fetch_read_chunk(state.connection);
        
        if (chunk == null) {
            // Download complete
            finish_download();
            return;
        }
        
        // Store chunk
        state.chunks.push(chunk);
        
        // Update progress
        var info = http.fetch_info(state.connection);
        state.downloaded_bytes = info.bytes_read;
        if (info.headers_complete && info.content_length > 0) {
            state.total_bytes = info.content_length;
        }
        
        // Schedule next chunk read
        $_.delay(read_next_chunk, 0);
        
    } catch (e) {
        // Error during download
        state.error = e.toString();
        if (state.connection) {
            http.fetch_close(state.connection);
        }
        
        if (state.download_msg) {
            send(state.download_msg, {
                type: 'error',
                error: state.error,
                url: state.current_url
            });
        }
        
        // Reset state
        state.downloading = false;
        state.connection = null;
        state.download_msg = null;
        state.chunks = [];
    }
}

// Complete the download and send result
function finish_download() {
    if (state.connection) {
        http.fetch_close(state.connection);
    }
    
    // Combine all chunks into single ArrayBuffer
    var total_size = 0;
    for (var i = 0; i < state.chunks.length; i++) {
        total_size += state.chunks[i].byteLength;
    }
    
    var result = new ArrayBuffer(total_size);
    var view = new Uint8Array(result);
    var offset = 0;
    
    for (var i = 0; i < state.chunks.length; i++) {
        var chunk_view = new Uint8Array(state.chunks[i]);
        view.set(chunk_view, offset);
        offset += state.chunks[i].byteLength;
    }
    
    // Send complete message
    if (state.download_msg) {
        send(state.download_msg, {
            type: 'complete',
            url: state.current_url,
            data: result,
            size: result.byteLength,
            duration: os.now() - state.start_time
        });
    }
    
    // Reset state
    state.downloading = false;
    state.connection = null;
    state.current_url = null;
    state.download_msg = null;
    state.chunks = [];
}