ismasan

# https://github.com/ismasan/parametric

require 'parametric'

class InputValidator
  def initialize(&block)
    @schema = Parametric::Schema.new(&block)
  end

  # @param result [Result]

ismasan

# A Pipeline extension to process steps concurrently
# Example
# class ConcurrentPipeline < Pipeline
#   include ConcurrentProcessing
# end
#
# MyPipeline = ConcurrentPipeline.new do |pl|
#   pl.step ValidateInput
#
#   # These steps run concurrently

ismasan

require 'json'
require 'csv'

# ruby json_to_csv.rb /path/to/data.json > /path/to/data.csv
#
MAP_VALUE = ->(value) {
  case value
  when Array
    value.join('|')
  when Hash

ismasan

module Hash64
  # Turn a string into a well distributed 64 bit integer
  # This matches the hash_64() PGSQL function used to partition streams.
  #
  # @param value [String] the string to hash
  # @return [Integer] a 64 bit integer
  def self.call(value)
    binary_string = ('0' + Digest::MD5.hexdigest(value)[0...16]).hex.to_s(2)
    # Convert the binary string to a signed BigInt
    bigint_value = binary_string.to_i(2)

ismasan

CREATE OR REPLACE FUNCTION public.rebalance_consumers_all_groups()
 RETURNS void
 LANGUAGE plpgsql
AS $function$
DECLARE
  group_ids uuid[];
  min_last_seq bigint;
  consumers_deleted bigint;
  group_min_seq bigint;
  p_group_id uuid;

ismasan

require 'digest'

class ConsistentHash
  Node = Data.define(:position, :updated_at)

  def initialize(size = 1)
    @hash_ring = {}
    (0...size).each { |i| add(i) }
  end

ismasan

require 'pg'
require 'connection_pool'
require 'async'

DB = ConnectionPool.new(size: 5, timeout: 5) do
  PG.connect(dbname: 'test_db')
end

class DBClient
  def initialize(db)

ismasan

# Turn a line-based Enumerator into an Enumerator
# that parses each CSV line and yields them as Hash rows
# Usage
#   csv = CSVStream.new(line_stream, headers: { product_name: 0, price: 2 })
#
# With a bit more work it could detect CSV headers automatically, from the first line.
require 'csv'

class CSVStream
  include Enumerable

ismasan

puts "START"
chan = Channel(Nil).new
spawn do
  sleep 6
  chan.send nil
end

select
when chan.receive
  puts "CHAN"

ismasan

# simple railway-style pipelines for Crystal

class Success(T)
  def initialize(@value : T)
  end

  def then(callable : T -> Success(T) | Failure(T))
    callable.call(@value)
  end