PHP code example of phunkie / streams

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

phunkie / streams example snippets

// Stream<Pure, Int>
$stream = Stream(1, 2, 3, 4);

// Pure streams can be converted to other collections
$stream->toList();   // List<Int> (1, 2, 3, 4)
$stream->toArray();  // [1, 2, 3, 4]

use const Phunkie\Functions\numbers\increment;

$stream = Stream(1, 2, 3, 4);

$stream->map(increment)->toList();
// List<Int> (2, 3, 4, 5)

Stream(1, 2, 3, 4)
    ->zipWith(increment)
    ->toList();
// List(Pair(1, 2), Pair(2, 3), Pair(3, 4), Pair(4, 5))

// Stream from a range
$fromRange = Stream(fromRange(0, 1000000000));
$fromRange->take(10)->compile()->toList();
// List(0, 1, 2, 3, 4, 5, 6, 7, 8, 9) 

// Stream from an iteration
$infiniteOdds = Stream(iterate(1)(fn ($x) => $x + 2));
$infiniteOdds->take(10)->compile()->toList();
// List(1, 3, 5, 7, 9, 11, 13, 15, 17, 19)

// Repeating a finite stream infinitely
$repeat = Stream(1, 2, 3)
    ->repeat()->take(12)->compile()->toList();
// List(1, 2, 3, 1, 2, 3, 1, 2, 3, 1, 2, 3)

Stream(1, 2, 3)
    ->repeat()
    ->runLog()
    ->unsafeRun();
// [1, 2, 3, 1, 2, 3, 1, 2, 3, ...] 

// Concatenation
Stream(1, 2, 3)
    ->concat(Stream(4, 5, 6))
    ->compile()
    ->toList();
// List(1, 2, 3, 4, 5, 6)

// Interleaving
$x = Stream(1, 2, 3, 4, 5);
$y = Stream("Monday", "Tuesday", "Wednesday", "Thursday", "Friday");
$z = Stream(true, false, true, false, true);

$x->interleave($y, $z)->compile()->toList();
// List(1, "Monday", true, 2, "Tuesday", false, 3, "Wednesday", true, 4,
// "Thursday", false, 5, "Friday", true)

port the necessary functions

// Create a stream
$stream = Stream(1, 2, 3, 4, 5);

// Process the stream
$result = $stream
    ->map(fn($x) => $x * 2)
    ->filter(fn($x) => $x > 5)
    ->toArray();

// Output: [6, 8, 10]
var_dump($result);

use Phunkie\Streams\IO\File\Path;
use function Phunkie\Streams\IO\File\{readFileContents, writeFileContents};

// Read file with automatic resource cleanup
$content = readFileContents(new Path('data.txt'))
    ->unsafeRunSync();

// Write file with guaranteed cleanup even on errors
$bytes = writeFileContents(new Path('output.txt'), "Hello, World!")
    ->unsafeRunSync();

use function Phunkie\Streams\IO\File\readFileContents;

// Using attempt() - returns Validation
$result = readFileContents(new Path('/nonexistent/file.txt'))
    ->attempt()
    ->unsafeRunSync();

$content = $result->getOrElse("default content");

// Using handleError() - recover from errors
$content = readFileContents(new Path('/nonexistent/file.txt'))
    ->handleError(fn($e) => "Error: " . $e->getMessage())
    ->unsafeRunSync();

use function Phunkie\Streams\IO\File\{readFileContents, writeFileContents};

$result = writeFileContents($path, "original")
    ->flatMap(fn($_) => readFileContents($path))
    ->map(fn($content) => strtoupper($content))
    ->flatMap(fn($upper) => writeFileContents($path, $upper))
    ->flatMap(fn($_) => readFileContents($path))
    ->unsafeRunSync();

// Result: "ORIGINAL"

$uppercase = fn(Stream $s) => $s->map(fn($x) => strtoupper($x));

$result = Stream(...['hello', 'world'])
    ->through($uppercase)
    ->toArray();
// ['HELLO', 'WORLD']

// Take elements while condition is true
Stream(...[1, 2, 3, 4, 5, 1, 2])
    ->takeWhile(fn($x) => $x < 4)
    ->toArray();
// [1, 2, 3]

// Drop elements while condition is true
Stream(...[1, 2, 3, 4, 5])
    ->dropWhile(fn($x) => $x < 3)
    ->toArray();
// [3, 4, 5]

Stream(...[1, 2, 3, 4, 5, 6])
    ->chunk(2)
    ->toArray();
// [[1, 2], [3, 4], [5, 6]]

use function Phunkie\Streams\IO\File\{writeFile, readLines};

// Write stream to file
Stream(...['line1', 'line2', 'line3'])
    ->through(writeFile(new Path('/tmp/output.txt')));

// With transformations
Stream(...[1, 2, 3, 4, 5])
    ->filter(fn($x) => $x % 2 === 0)
    ->map(fn($x) => "Even: $x")
    ->through(writeFile(new Path('/tmp/evens.txt')));

// Complex pipeline
$processData = fn(Stream $s) => $s
    ->dropWhile(fn($x) => $x < 5)
    ->takeWhile(fn($x) => $x <= 15)
    ->filter(fn($x) => $x % 2 === 0)
    ->map(fn($x) => "Value: $x");

Stream(...range(1, 20))
    ->through($processData)
    ->through(writeFile(new Path('/tmp/processed.txt')));

use Phunkie\Streams\Network;

// HTTP GET
$data = Network::httpGet('https://api.example.com/data')
    ->compile->toArray();

// HTTP POST with JSON
Network::httpPost(
    'https://api.example.com/users',
    json_encode(['name' => 'Alice']),
    ['Content-Type: application/json']
)->compile->toArray();

// Stream processing
Network::httpGet('https://api.example.com/stream')
    ->map(fn($chunk) => json_decode($chunk, true))
    ->filter(fn($data) => $data !== null)
    ->compile->toArray();

use Phunkie\Streams\{Network, IO\Network\SocketAddress};

Network::client(new SocketAddress('localhost', 8080))
    ->map(fn($data) => processData($data))
    ->compile->toArray();

Network::server(host: 'localhost', port: 8080)
    ->map(function($client) {
        $data = fread($client, 1024);
        fwrite($client, "Echo: $data");
        fclose($client);
        return "Handled client";
    })
    ->take(10)
    ->compile->drain;

Stream(...['message1', 'message2', 'message3'])
    ->through(Network::socketWrite(
        new SocketAddress('localhost', 8080)
    ));

use Phunkie\Streams\Type\Stream;

// Process elements in parallel (max 4 concurrent)
Stream(1, 2, 3, 4, 5, 6, 7, 8)
    ->parMap(4, fn($x) => expensiveComputation($x))
    ->compile()
    ->toArray();

// Parallel with error collection (doesn't fail fast)
Stream(1, 2, 3, 4, 5)
    ->parMapValidation(2, fn($x) => riskyOperation($x))
    ->compile()
    ->toArray();
// Returns: [Success(1), Failure($e), Success(3), ...]

use Phunkie\Streams\Network;
use Phunkie\Effect\IO\IO;

// Process IO effects in parallel
Stream("url1", "url2", "url3")
    ->parEvalMap(2, fn($url) => Network::httpGet($url))
    ->compile()
    ->drain
    ->unsafeRunSync();

// Deferred parallel execution
$io = Stream("url1", "url2", "url3")
    ->parTraverse(2, fn($url) => Network::httpGet($url));

$results = $io->unsafeRunSync(); // Stream of results

// Merge multiple streams concurrently
$stream1 = Stream(1, 2, 3);
$stream2 = Stream(4, 5, 6);
$stream3 = Stream(7, 8, 9);

Stream::parMerge($stream1, $stream2, $stream3)
    ->compile()
    ->toArray();

// Concurrent flatMap
Stream("user1", "user2", "user3")
    ->parMergeMap(2, fn($user) =>
        Stream($user->getPosts())
    )
    ->compile()
    ->toArray(); // All posts from all users

use function Phunkie\Streams\IO\File\writeFile;

// Process a 10GB log file with constant ~4MB memory usage
Stream(new Path('huge-10gb-log.txt'))
    ->filter(fn($line) => str_contains($line, 'ERROR'))
    ->map(fn($line) => processLine($line))
    ->through(writeFile(new Path('errors.txt')));

// Process millions of records with bounded memory
Stream::fromLargeDataset()
    ->parMap(4, fn($record) => processRecord($record))
    ->chunk(1000)
    ->through(writeFile(new Path('output.txt')));