Pipes and Filters

void main() {
  final data = ['  hello', 'world', 'Hello', '  dart', 'world'];

  final result = pipeline(
    data,
    [
      upperCaseFilter,
      trimFilter,
      deduplicateFilter,
    ],
  );

  print(result); // Output: [HELLO, WORLD, DART]
}

typedef StringFilter = String Function(String);

List<String> pipeline(List<String> data, List<StringFilter> filters) {
  return data.map((item) => filters.fold(item, (String current, StringFilter filter) => filter(current))).toList();
}

String upperCaseFilter(String input) {
  return input.toUpperCase();
}

String trimFilter(String input) {
  return input.trim();
}

String deduplicateFilter(String input) {
  // In a real-world scenario, you'd likely use a Set
  // to track seen strings and only return the first occurrence.
  // This is a simplified example.
  return input;
}

// PipesAndFilters.scala
object PipesAndFilters {

  // Define a simple filter trait
  trait Filter[T, U] {
    def apply(data: T): U
  }

  // A source of data
  val source = "This is a test. This is only a test."

  // Define filters
  val splitWords: Filter[String, Array[String]] = (text: String) => text.split("\\s+").map(_.toLowerCase)
  val filterShortWords: Filter[Array[String], Array[String]] = (words: Array[String]) => words.filter(_.length > 2)
  val countWords: Filter[Array[String], Map[String, Int]] = (words: Array[String]) =>
    words.groupBy(identity).mapValues(_.length)

  // The pipeline
  val result: Map[String, Int] = source
    .split("\\s+")
    .map(_.toLowerCase)
    .filter(_.length > 2)
    .groupBy(identity)
    .mapValues(_.length)

    def main(args: Array[String]): Unit = {
      println(result)
    }
}

<?php

/**
 * Filter 1: Uppercase the input string
 */
function uppercaseFilter(iterable $input): iterable
{
    foreach ($input as $item) {
        yield strtoupper($item);
    }
}

/**
 * Filter 2: Remove spaces from the input string
 */
function removeSpacesFilter(iterable $input): iterable
{
    foreach ($input as $item) {
        yield str_replace(' ', '', $item);
    }
}

/**
 * Filter 3: Add a prefix to each string.
 */
function addPrefixFilter(iterable $input, string $prefix): iterable
{
    foreach ($input as $item) {
        yield $prefix . $item;
    }
}


// Initial data
$data = ["hello world", "php is fun"];

// Create the pipeline
$processedData = uppercaseFilter($data);
$processedData = removeSpacesFilter($processedData);
$processedData = addPrefixFilter($processedData, 'Processed: ');

// Consume the pipeline
foreach ($processedData as $item) {
    echo $item . PHP_EOL;
}

?>

# Pipes and Filters Pattern in Ruby

def extract_words(text)
  text.downcase.scan(/\b\w+\b/)
end

def filter_long_words(words, min_length)
  words.select { |word| word.length >= min_length }
end

def count_words(words)
  words.tally
end

text = "This is a Sample Text to demonstrate Pipes and Filters."

word_counts = text.to_s
                  .then { |t| extract_words(t) }
                  .then { |w| filter_long_words(w, 3) }
                  .then { |f| count_words(f) }

puts word_counts

protocol Filter {
    func execute(input: String) -> String
}

struct StringToUppercaseFilter: Filter {
    func execute(input: String) -> String {
        input.uppercased()
    }
}

struct RemoveWhitespaceFilter: Filter {
    func execute(input: String) -> String {
        input.trimmingCharacters(in: .whitespacesAndNewlines)
    }
}

struct ExclaimifyFilter: Filter {
    func execute(input: String) -> String {
        input + "!"
    }
}

class FilterChain {
    private let filters: [Filter]

    init(filters: [Filter]) {
        self.filters = filters
    }

    func execute(input: String) -> String {
        var result = input
        for filter in filters {
            result = filter.execute(input: result)
        }
        return result
    }
}

// Example Usage:
let filters = [
    StringToUppercaseFilter(),
    RemoveWhitespaceFilter(),
    ExclaimifyFilter()
]

let chain = FilterChain(filters: filters)
let finalResult = chain.execute(input: "  hello world ")

print(finalResult) // Output: HELLO WORLD!

// Pipes and Filters Pattern in Kotlin

data class DataItem(val value: String)

// Filter 1: Uppercase the string
fun filterUppercase(data: List<DataItem>): List<DataItem> = data.map { it.copy(value = it.value.uppercase()) }

// Filter 2: Remove characters that are not letters or numbers
fun filterAlphanumeric(data: List<DataItem>): List<DataItem> = data.map { it.copy(value = it.value.filter { char -> char.isLetterOrDigit() }) }

// Filter 3: Filter out empty strings
fun filterNotEmpty(data: List<DataItem>): List<DataItem> = data.filter { it.value.isNotEmpty() }

fun main() {
    val inputData = listOf(
        DataItem(" Hello World! "),
        DataItem("123abc456"),
        DataItem(""),
        DataItem("Kotlin Pipes")
    )

    val processedData = inputData
        .let(filterUppercase)
        .let(filterAlphanumeric)
        .let(filterNotEmpty)
    
    println(processedData) // Output: [DataItem(value=HELLO WORLD!), DataItem(value=123ABC456), DataItem(value=KOTLINPIPES)]
}

use std::sync::mpsc;
use std::thread;

fn main() {
    // Create channels
    let (tx1, rx1) = mpsc::channel();
    let (tx2, rx2) = mpsc::channel();
    let (tx3, rx3) = mpsc::channel();

    // Input data
    let input_data = vec![1, 2, 3, 4, 5];

    // Spawn threads for filters
    let handle1 = thread::spawn(move || {
        for item in input_data {
            tx1.send(item * 2).unwrap();
        }
        drop(tx1); // Signal end of stream
    });

    let handle2 = thread::spawn(move || {
        for item in rx1 {
            tx2.send(item.to_string()).unwrap();
        }
        drop(tx2);
    });

    let handle3 = thread::spawn(move || {
        for item in rx2 {
            tx3.send(format!("Result: {}", item)).unwrap();
        }
        drop(tx3);
    });

    // Process the output
    for item in rx3 {
        println!("{}", item);
    }

    // Wait for threads to finish
    handle1.join().unwrap();
    handle2.join().unwrap();
    handle3.join().unwrap();
}

// pipeline.go

package main

import "fmt"

// source generates strings and sends them to the channel.
func source(out chan<- string) {
	defer close(out)
	strings := []string{"hello", "world", "go", "pipeline"}
	for _, s := range strings {
		out <- s
	}
}

// filter converts strings to uppercase and sends them to the channel.
func filter(in <-chan string, out chan<- string) {
	defer close(out)
	for s := range in {
		out <- fmt.Sprintf("%s", s) // Convert to uppercase
	}
}

// sink receives strings from the channel and prints them.
func sink(in <-chan string) {
	for s := range in {
		fmt.Println(s)
	}
}

func main() {
	// Create channels
	ch1 := make(chan string)
	ch2 := make(chan string)

	// Start goroutines for each stage of the pipeline
	go source(ch1)
	go filter(ch1, ch2)
	go sink(ch2)

	// Keep the main goroutine alive to allow the pipeline to complete
	select {}
}

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <ctype.h>

// Filter 1: Uppercase filter
void uppercase_filter(FILE *in, FILE *out) {
    int c;
    while ((c = fgetc(in)) != EOF) {
        fputc(toupper(c), out);
    }
}

// Filter 2: Remove spaces filter
void remove_spaces_filter(FILE *in, FILE *out) {
    int c;
    while ((c = fgetc(in)) != EOF) {
        if (c != ' ') {
            fputc(c, out);
        }
    }
}

int main() {
    int pipe1[2];
    int pipe2[2];
    pid_t pid1, pid2;

    if (pipe(pipe1) == -1 || pipe(pipe2) == -1) {
        perror("pipe");
        exit(EXIT_FAILURE);
    }

    pid1 = fork();
    if (pid1 == -1) {
        perror("fork");
        exit(EXIT_FAILURE);
    }

    if (pid1 == 0) { // Child 1: Uppercase filter
        dup2(pipe1[0], STDIN_FILENO);
        dup2(pipe1[1], STDOUT_FILENO);
        close(pipe1[0]);
        close(pipe1[1]);
        close(pipe2[0]);
        close(pipe2[1]);
        uppercase_filter(stdin, stdout);
        exit(EXIT_SUCCESS);
    }

    pid2 = fork();
    if (pid2 == -1) {
        perror("fork");
        exit(EXIT_FAILURE);
    }

    if (pid2 == 0) { // Child 2: Remove spaces filter
        dup2(pipe1[1], STDIN_FILENO);
        dup2(pipe2[1], STDOUT_FILENO);
        close(pipe1[0]);
        close(pipe1[1]);
        close(pipe2[0]);
        close(pipe2[1]);
        remove_spaces_filter(stdin, stdout);
        exit(EXIT_SUCCESS);
    }

    // Parent: Connect to the pipeline and read the output
    close(pipe1[0]);
    close(pipe1[1]);
    close(pipe2[1]);

    FILE *output = fdopen(pipe2[0], "r");
    if (output == NULL) {
        perror("fdopen");
        exit(EXIT_FAILURE);
    }

    char buffer[100];
    printf("Output: ");
    while (fgets(buffer, sizeof(buffer), output) != NULL) {
        printf("%s", buffer);
    }

    fclose(output);
    wait(NULL);
    wait(NULL);

    return 0;
}

#include <iostream>
#include <string>
#include <algorithm>
#include <functional>

// Filters
std::string to_uppercase(const std::string& input) {
  std::string result = input;
  std::transform(result.begin(), result.end(), result.begin(), ::toupper);
  return result;
}

std::string remove_spaces(const std::string& input) {
  std::string result;
  std::remove_copy_if(input.begin(), input.end(), std::back_inserter(result), ::isspace);
  return result;
}

bool is_palindrome(const std::string& input) {
  std::string reversed_input = input;
  std::reverse(reversed_input.begin(), reversed_input.end());
  return input == reversed_input;
}

int main() {
  std::string input = "Race car";

  // Define the processing pipeline
  auto pipeline = [&to_uppercase, &remove_spaces, &is_palindrome](const std::string& input_str) {
    std::string stage1 = to_uppercase(input_str);
    std::string stage2 = remove_spaces(stage1);
    return is_palindrome(stage2);
  };
  
  // Execute the pipeline
  bool result = pipeline(input);
  std::cout << "Is '" << input << "' a palindrome? " << (result ? "Yes" : "No") << std::endl;

  return 0;
}

using System;
using System.Collections.Generic;
using System.Linq;

public static class Pipeline
{
    public static T Pipe<T>(T input, params Func<T, T>[] filters)
    {
        return filters.Aggregate(input, (acc, filter) => filter(acc));
    }
}

public class Example
{
    public static void Main(string[] args)
    {
        string inputString = "  Hello, World!  ";

        // Define filters
        Func<string, string> trimFilter = s => s.Trim();
        Func<string, string> toLowerFilter = s => s.ToLower();
        Func<string, string> replaceFilter = s => s.Replace("world", "c#");

        // Create and execute the pipeline
        string result = Pipeline.Pipe(inputString, trimFilter, toLowerFilter, replaceFilter);

        Console.WriteLine($"Original: '{inputString}'");
        Console.WriteLine($"Processed: '{result}'");
    }
}

// Define filter functions
const toUpperCaseFilter = (input: string): string => input.toUpperCase();
const trimWhitespaceFilter = (input: string): string => input.trim();
const replaceCommasFilter = (input: string): string => input.replace(/,/g, ".");

// Pipe function to compose filters
const pipe = <T>(initialValue: T, ...filters: ((input: T) => T)[]): T => {
  return filters.reduce((value, filter) => filter(value), initialValue);
};

// Example usage
const originalString = " hello, world ";
const transformedString = pipe(originalString, toUpperCaseFilter, trimWhitespaceFilter, replaceCommasFilter);

console.log(`Original: "${originalString}"`);
console.log(`Transformed: "${transformedString}"`);

// Filter: Uppercase Filter
const uppercaseFilter = (input) => input.toUpperCase();

// Filter: Remove Whitespace Filter
const removeWhitespaceFilter = (input) => input.trim();

// Filter: Split into words Filter
const splitIntoWordsFilter = (input) => input.split(' ');

// Filter: Filter words longer than 3 characters
const longWordFilter = (words) => words.filter(word => word.length > 3);

// Pipe (Chain) function
const pipe = (data, ...filters) => filters.reduce((result, filter) => filter(result), data);

// Example Usage
const rawData = '   hello world this is a test   ';

const processedData = pipe(rawData,
  removeWhitespaceFilter,
  uppercaseFilter,
  splitIntoWordsFilter,
  longWordFilter
);

console.log(processedData); // Output: ["HELLO", "WORLD", "THIS", "TEST"]

def load_data(filename):
    """Filter 1: Loads data from a file (simulated here)."""
    with open(filename, 'r') as f:
        for line in f:
            yield line.strip()

def filter_long_lines(data, max_length):
    """Filter 2: Filters lines longer than max_length."""
    for line in data:
        if len(line) <= max_length:
            yield line

def uppercase_lines(data):
    """Filter 3: Converts lines to uppercase."""
    for line in data:
        yield line.upper()

def remove_duplicates(data):
    """Filter 4: Removes duplicate lines."""
    seen = set()
    for line in data:
        if line not in seen:
            yield line
            seen.add(line)

def main():
    """Creates and runs the pipe and filter pipeline."""
    filename = 'data.txt'
    # Create a dummy data file
    with open(filename, 'w') as f:
        f.write("apple\n")
        f.write("banana\n")
        f.write("orange\n")
        f.write("apple\n")
        f.write("kiwi long line\n")
        f.write("grape\n")

    processed_data = remove_duplicates(
        uppercase_lines(
            filter_long_lines(
                load_data(filename),
                10
            )
        )
    )

    for line in processed_data:
        print(line)

if __name__ == "__main__":
    main()

import java.util.Arrays;
import java.util.stream.Stream;

interface Filter {
    String process(String input);
}

class UpperCaseFilter implements Filter {
    @Override
    public String process(String input) {
        return input.toUpperCase();
    }
}

class RemoveSpacesFilter implements Filter {
    @Override
    public String process(String input) {
        return input.replaceAll("\\s+", "");
    }
}

class Pipeline {
    private final Stream<Filter> filters;

    public Pipeline(Filter... filters) {
        this.filters = Arrays.stream(filters);
    }

    public String run(String input) {
        return filters.map(filter -> filter.process(input))
                .reduce(input, (acc, filterResult) -> filterResult)
                .orElse(input); // handles empty filter stream
    }
}

public class PipesAndFilters {
    public static void main(String[] args) {
        Pipeline pipeline = new Pipeline(new UpperCaseFilter(), new RemoveSpacesFilter());
        String input = "  Hello world!  ";
        String result = pipeline.run(input);
        System.out.println("Input: " + input);
        System.out.println("Result: " + result);
    }
}