Download the PHP package phunkie/streams without Composer

Phunkie Streams

Phunkie Streams is a PHP functional library for working with streams inspired by functional streaming libraries like fs2 (Scala). It allows you to process data in a declarative, composable way.

Installation

Features

• Pure streams: Finite sequences of values that can be transformed and combined
• Infinite streams: Unbounded sequences that can be processed lazily
• Effectful operations: Operations that interact with the outside world (I/O, etc.) using phunkie/effect
• Resource management: Safe handling of resources through bracket pattern for guaranteed cleanup
• Error handling: Functional error handling with attempt() and handleError()
• Monadic composition: Compose IO operations with flatMap() for type-safe pipelines
• Stream operations: Rich set of operations including through(), takeWhile(), dropWhile(), chunk()
• Memory efficient: True streaming with constant memory usage regardless of data size
• Parallel processing: Concurrent operations with automatic fallback and memory optimization

What's Implemented

Phunkie Streams has completed Phases 1-3 of development. Here's what's currently available:

Core Stream Types

• Pure Streams - Stream(1, 2, 3) for finite in-memory sequences
• Infinite Streams - Stream(iterate(...)), Stream(fromRange(...)) for unbounded sequences
• Resource Streams - File and network I/O streams with automatic resource management

Stream Operations

• Transformations: map(), filter(), flatMap()
• Composition: concat(), merge(), interleave(), zip(), zipWith()
• Control flow: take(), drop(), takeWhile(), dropWhile()
• Batching: chunk() for processing in fixed-size chunks
• Pipes: through() for composable transformation pipelines
• Compilation: compile()->toArray(), compile()->toList(), compile->drain

File I/O

• Reading: Stream(new Path('file.txt')), readFileContents(), readLines()
• Writing: writeFileContents(), writeLines(), writeFile() pipe function
• Utilities: exists(), deleteFile()
• All file operations use bracket() internally for guaranteed cleanup

Network Operations

• HTTP: Network::httpGet(), Network::httpPost(), Network::httpPut(), Network::httpDelete()
• TCP Client: Network::client(SocketAddress) for socket connections
• TCP Server: Network::server(host:, port:) for accepting connections
• Socket Writing: Network::socketWrite() pipe function
• All network resources use automatic cleanup via __destruct()

Resource Management

• Bracket pattern: bracket(acquire, use, release) from phunkie/effect
• Automatic cleanup: Resource objects (HttpRequest, SocketRead, etc.) clean up automatically
• Safe by default: All provided file/network functions handle resources safely
• See Resource Management Guide for patterns

Error Handling

• Validation: IO->attempt() converts exceptions to Validation<Throwable, A>
• Recovery: IO->handleError(fn) for error recovery with fallback values
• Composition: Chain error handling with flatMap() for complex scenarios
• See Error Handling Guide for comprehensive patterns

Concurrency & Parallel Processing

• Parallel operations: parMap(), parMapValidation(), parEvalMap(), parTraverse(), parEval()
• Concurrent merging: parMerge(), parMergeMap() for parallel stream combining
• Auto CPU detection: Automatically detects CPU cores for optimal parallelism
• Automatic fallback: Falls back to sequential execution on concurrency failures
• Memory optimized: All parallel operations process incrementally without materializing entire streams

Not Yet Implemented

• Backpressure: Flow control mechanisms (planned for future release)
• Connection pooling: Resource pooling for HTTP/sockets (planned for future release)
• Process integration: System command execution (future consideration)

Pure Streams

Transformations on Pure Streams

Infinite Streams

Picking on Infinite Streams with runlog

Combining Streams

Basic Usage

Resource Management with Bracket

Phunkie Streams uses the bracket pattern from phunkie/effect for safe resource management:

See examples/bracket.php for more examples.

Error Handling

Functional error handling with attempt() and handleError():

See doc/error-handling.md for comprehensive patterns.

Stream Composition with flatMap

Compose IO operations in type-safe pipelines:

See doc/composition.md for detailed patterns.

Stream Operations

through() - Pipe Operator

Apply transformation pipelines to streams:

takeWhile() and dropWhile()

chunk() - Batch Processing

See examples/stream-operations.php for 20 comprehensive examples.

File I/O with Streams

writeFile() - Stream Pipe for Writing

Write stream elements directly to files using the writeFile() pipe:

See examples/file-pipes.php for 12 comprehensive file I/O examples.

Network Operations

HTTP Requests

Make HTTP requests as streams:

TCP Sockets

TCP Client

TCP Server

Writing to Sockets

See examples/network.php for 15 comprehensive network examples.

Concurrency & Parallel Processing

Phunkie Streams provides concurrent stream processing with automatic fallback to sequential execution:

Parallel Map Operations

Parallel IO Effects

Parallel Stream Merging

Memory Efficiency

All parallel operations are memory-optimized to process data incrementally:

• Auto-detects CPU cores when maxConcurrent = 0
• Processes elements in chunks without materializing entire streams
• Automatic fallback to sequential execution on concurrency failures
• Constant memory usage regardless of stream size

Memory Optimization

Phunkie Streams is designed for constant memory usage regardless of data size. All operations use true streaming with lazy evaluation:

File Operations

The writeFile() pipe function processes streams incrementally without loading everything into memory:

Parallel Operations

All parallel operations (parMap, parEvalMap, parTraverse, parMerge, parMergeMap) process data in chunks without materializing entire streams:

Iterator Protocol

Operations use PHP's Iterator protocol for memory-efficient streaming:

• Elements are pulled on-demand, one at a time
• Transformations (map, filter) are applied per-element
• No intermediate arrays are created unnecessarily
• Memory usage stays constant regardless of input size

Best Practices

1. Avoid toArray() on large streams - Use compile->drain for side effects
2. Use chunk() for batching - Process data in manageable batches
3. Leverage through() - Compose pipes for reusable transformations
4. Use parallel operations - Let the library handle concurrent processing efficiently

Documentation

• Resource Management Guide - When to use bracket() vs __destruct()
• Error Handling Guide - Error recovery strategies
• Composition Guide - Monadic composition patterns
• Full Documentation - Complete documentation directory

Examples

• examples/bracket.php - Resource management (10 examples)
• examples/error-handling.php - Error handling (12 examples)
• examples/composition.php - Stream composition (12 examples)
• examples/stream-operations.php - Stream operations (20 examples)
• examples/file-pipes.php - File I/O with streams (12 examples)
• examples/network.php - Network operations (15 examples)

Contributing

We welcome contributions to Phunkie Streams! Please see our Contributing Guide for more information.

License

Phunkie Streams is licensed under the LICENSE file included in the repository.