Pipe Dreams

pipe.rb

class Pipe
  def self.unwrap
    self
  end

  def self._unwrap(context, input, ops)
    ops.inject(input) { |value, op| context.send(op, value) }
  end

  def initialize(context, input)
    @context = context
    @input = input
    @ops = []
  end

  def add_op(name)
    if name == Pipe
      self.class._unwrap(@context, @input, @ops)
    else
      @ops << name
      self
    end
  end
  
  alias >> add_op
end

def Pipe(input)
  Pipe.new(self, input)
end

piping.rb

require 'ostruct'

class Example
  def self.call
    new.call
  end
  
  def call
    Pipe(data) >>
      :group >>
      :map >>
      :sort >>
      Pipe.unwrap
  end
  
  private
  
  def data
    [
      OpenStruct.new(id: 1, name: 'Jack', value:  1),
      OpenStruct.new(id: 2, name: 'Jill', value: 10),
      OpenStruct.new(id: 3, name: 'Jack', value:  5),
    ]
  end
  
  def group(ungrouped) 
    ungrouped.each_with_object(
      Hash.new { |h, k| h[k] = [] }
    ) do |item, groups|
      groups[item.name] << item
    end
  end
  
  def map(group_map)
    group_map.map { |name, group|
      { name: name, balance: group.sum(&:value) }
    }
  end
  
  def sort(balances)
    balances.sort { |a, b| b[:balance] <=> a[:balance] }
  end
end

Good stuff!