From: Matsuo Hisanori <hisanori@...>
Date: 2000-02-28T12:09:32+09:00
Subject: [ruby-list:21069] Hash を RDBMS のインタフェイスに (Re:  Re: Hash#update 	について)

This is a multi-part message in MIME format.
--------------E8723A67EE91239713117D17
Content-Type: text/plain; charset=iso-2022-jp
Content-Transfer-Encoding: 7bit

松尾です。

# joinやupdateな文脈とは関係なく出てきてます_o_

"NAKAMURA, Hiroshi" wrote:

> HashをRDBMSにおけるロウのインスタンスとして持つような
> インタフェイスがあれば，

僕の誤解でなければ、そんなライブラリを作って使ってます。御興味がある方は
御覧ください。

必要なもの

Ruby 1.5系(1.4系だと駄目。observer.rbを1.5系のものにすればOK)
PostgreSQL
PostgreSQL拡張ライブラリ
    (ftp://ftp.netlab.co.jp/pub/lang/ruby/contrib/postgres-0.5.3.tar.gz)
application.rb
    (http://www.jin.gr.jp/~nahi/Ruby/ruby.shtml#application)

# 残念ながらrdbcではなく、PostgreSQL拡張ライブラリを直打ちしてます_o_
# その辺は何れ御相談させてください。


機能はこんな感じです。

全ての入出力をハッシュに限定した、RDBのラッパ
  1. エンティティクラスの宣言
  2. エンティティホームインスタンスの生成

エンティティホーム
  1. エンティティの登録 insert
  2. エンティティの更新 update
  3. エンティティの削除 delete
  4. エンティティの選択 select(とfind)
  5. セレクタの追加
  6. キャッシュの準備とキャッシュセレクタの追加

トランザクション管理
--------------E8723A67EE91239713117D17
Content-Type: text/plain; charset=iso-2022-jp;
 name="Entity.rb"
Content-Transfer-Encoding: 8bit
Content-Disposition: inline;
 filename="Entity.rb"

# $Id: Entity.rb,v 1.1 2000/02/23 16:33:31 matsuo Exp $

# 全ての入出力をハッシュに限定した、RDBのラッパ
# method defineEntityType エンティティタイプ宣言メソッド
# 機能
#   1. エンティティクラスの宣言
#   2. エンティティホームインスタンスの生成

# class EntityHome テーブルのラッパ
# 機能
# 1. エンティティの登録 insert
#   insertEntity(entity) -> entity
#   
# 2. エンティティの更新 update
#   updateEntity(entity) -> entity
#   
# 3. エンティティの削除 delete
#   deleteEntity(key) -> key
#   
# 4. エンティティの選択 select
#   enumeration = select(selectorKey,key,...)
#   enumeration = select(selectorKey,Hash)
#   entity = find(selectorKey,key,...)
#   entity = find(selectorKey,Hash)
#
# 5. DBセレクタの追加
#   addSelector(selectorKey,parametorTypes,sql)
#
#   以下の二つの選択手段は予め用意されている
#     プライマリキーによる検索(セレクタキーは'pk')
#     全部(セレクタキーは'all')
#
# 6. キャッシュの準備とキャッシュセレクタの追加
#   prepareCacher
#    プライマリキーによる検索(セレクタキーは'pk')が自動的に追加される
#   addCacher(selectorKey,parametorTypes,sql)
#

# 2種類のセレクタ
# class EntitySelector : SQLでDBを検索する
# class EntityCacher : 与えられたキーとの完全一致でキャッシュを検索
# 使い方
#   enumeration = select(Array|Hash)
#   entity = find(Array|Hash)


require 'observer'
require 'thread'

module Entity; end # 単に宣言してるだけ。
def Entity.define(entityType,fieldType,pkName,tableName); eval(<<END,binding); end
class #{entityType} < Hash
  include Comparable

  EntityType =  #{entityType}
  FieldType = {#{fieldType.collect{|k,v| "'#{k}' => #{v.to_s}"}.join(', ')}}
  PkNames = [#{pkName.collect{|i| "'#{i}'"}.join(', ')}]
  TableName = '#{tableName}'
  
  def #{entityType}.entityType; return EntityType; end
  def #{entityType}.fieldType;  return FieldType;  end
  def #{entityType}.pkNames;    return PkNames;    end
  def #{entityType}.tableName;  return TableName;  end
  
  def #{entityType}.toEntity(aHash)
    return EntityType.new.update(aHash)
  end
  
  def #{entityType}.toRecord(aHash)
    record = {}
    
    aHash.each{|k,v|
      if FieldType[k] == Integer then
	record[k] = v
	
      elsif FieldType[k] == String then
	record[k] = v
	
      elsif FieldType[k] == Time then
	record[k] = v.to_i
	
      end
    }
    
    return record
  end

  def <=>(other)
    PkNames.each{|name|
      if self[name] < other[name] then
	return -1
      elsif self[name] > other[name] then
	return 1
      end
    }
    
    return 0
  end
end

#{entityType}Home = EntityHome.new(#{entityType})

LogDev.add(SEV_INFO,"#{entityType} defined.",'Entity')
LogDev.add(SEV_INFO,"#{entityType}Home defined.",'EntityHome')
END


class EntitySelector
  def initialize(entityType,parametorTypes,sql) # parametorTypes : パラメータの型のArray
    @entityType = entityType
    @parametorTypes = parametorTypes
    @selectSQL = "select #{sql}"
  end

  def select(*arg)
    if arg.size == 1 && arg[0].is_a?(Hash) then
      parametor = arg[0]
      arg = []
      @entityType.pkNames.each{|name| arg << parametor[name]}
    end

    #p @selectSQL

    record = toRecord(arg)

    statement = @selectSQL.dup
    record.each_index{|i|
      statement.gsub!(/:#{i.to_s}/,"'#{record[i].to_s}'")
    }

    return toEntityArray(Transaction.getConnection.exec(statement))
  end

  def find(*arg)
    return select(*arg)[0]
  end

  private

  def toRecord(arg)
    record = []

    arg.each_index{|i|
      if @parametorTypes[i] == Integer then
	record << arg[i]

      elsif @parametorTypes[i] == String then
	record << arg[i]

      elsif @parametorTypes[i] == Time then
	record << arg[i].to_i

      end
    }
  end

  def toEntityArray(pgResult)
    if pgResult then
      result = pgResult.result
      fields = pgResult.fields

      result.filter{|r|
	toEntity(r,fields)
      }
    else
      []
    end
  end

  def toEntity(record,fields)
    entity = @entityType.new

    for i in 0 .. fields.size - 1

      name = fields[i]
      value = record[i].taint # データベース中の値は常に汚れている。
                              # 汚れてないフィールドのサポートは将来に。

      if @entityType.fieldType[name] == Integer then
	entity[name] = value.to_i
      elsif @entityType.fieldType[name] == String then
	entity[name] = value.strip
      elsif @entityType.fieldType[name] == Time
	entity[name] = Time.at(value.to_i)
      end
    end

    return entity
  end
end


class EntityCacher
  attr('cache')

  # newはトランザクションのコンテキストで呼ばなければならない
  # entityHome : 検索対象のエンティティホーム
  # fieldNames : 検索キーとなるフィールドの名前のリスト
  def initialize(entityHome,fieldNames)
    @entityHome = entityHome
    @fieldNames = fieldNames

    @selfKey = id.to_s
    @m = Mutex.new
    @cache = {}

    @entityHome.select('all').each{|entity|
      key = makeKey(entity)
      if @cache[key] then
	@cache[key] << entity
      else
	@cache[key] = [entity]
      end
    }

    @entityHome.add_observer(self)
  end

  def select(*arg)
    if arg.size == 1 && arg[0].is_a?(Hash) then
      parametor = arg[0]
      arg = []
      @entityHome.entityType.pkNames.each{|name| arg << parametor[name]}
    end

    theCache = getTransactionCache

#    touch(arg,theCache)
#
#    if target = theCache[arg] then
#      return target.dup # リストは大切で、要素は大切じゃないから浅いコピー
#    else
#      return []
#    end

    if (target = theCache[arg]) then
      return target.dup # リストは大切で、要素は大切じゃないから浅いコピー
    elsif (target = @cache[arg]) then
      return target.collect{|e| e.dup} # リストも要素も大切だから深いコピー
    else
      return []
    end
  end

  def find(*arg)
    if arg.size == 1 && arg[0].is_a?(Hash) then
      parametor = arg[0]
      arg = []
      @entityHome.entityType.pkNames.each{|name| arg << parametor[name]}
    end

    theCache = getTransactionCache

#    touch(arg,theCache)
#
#    if target = theCache[arg] then
#      return target[0] # 要素は大切じゃないからそのまま。
#    else
#      return nil
#    end

    if (target = theCache[arg]) then
      return target[0] # 要素は大切じゃないからそのまま。
    elsif (target = @cache[arg]) then
      return target[0].dup # 要素は大切。だからdup。
    else
      nil
    end
  end

  def update(*arg)
    case arg[0]
    when 'commit','rollback'
      updateTransaction(*arg)
    when 'insert','update','delete'
      updateDB(*arg)
    end
  end

  private
  
  def updateDB(*arg)
    theCache = getTransactionCache

    case arg[0]
    when 'insert'
      insert(arg[1],theCache)

    when 'update'
      delete(arg[2],theCache)
      insert(arg[1],theCache)
      
    when 'delete'
      delete(arg[2],theCache)

    end
  end

  def getTransactionCache
    transaction = Transaction.current
    caches = transaction.caches

    unless caches[@selfKey] then
      caches[@selfKey] = {}
      transaction.add_observer(self)
    end

    return caches[@selfKey]
  end

  def touch(key,theCache)
    unless theCache[key] then
      theCache[key] = []
      if @cache[key] then
	@cache[key].each{|e| theCache[key] << e.dup }
      end
    end
  end

  def insert(entity,theCache)
    key = makeKey(entity)
    touch(key,theCache)
    theCache[key] << entity.dup
  end

  def delete(entity,theCache)
    key = makeKey(entity)
    touch(key,theCache)
    theCache[key].delete_if{|e| e == entity}
  end

  def updateTransaction(*arg)
    transaction = Transaction.current
    caches = transaction.caches

    unless caches[@selfKey] then
      caches[@selfKey] = {}
    end

    case arg[0]
    when 'commit'
      @m.synchronize {
	caches[@selfKey].each{|k,v|
	  if v.empty? then
	    @cache[k] = nil
	  else
	    @cache[k] = v
	  end
	}
      }

    when 'rollback'
    end

    caches[@selfKey] = nil
    transaction.delete_observer(self)
  end

  def makeKey(entity)
    key = []
    @fieldNames.each{|i|
      key << entity[i]
    }
    return key
  end
end

class EntityHome
  include Observable

  srand
  IdBase = 2 ** 29
#  IdBase = 2 ** 3

  attr('entityType')

  def initialize(entityType)
    @entityType = entityType

    @selector = {}
    @cached = false

    @insertSQL = <<END
insert into #{@entityType.tableName}
(#{@entityType.fieldType.keys.join(', ')})
values (#{@entityType.fieldType.keys.collect{|i| ':' + i}.join(', ')});
END

    @updateSQL = <<END
update #{@entityType.tableName}
set #{@entityType.fieldType.keys.collect{|i| i + ' = :' + i}.join(', ')}
where #{@entityType.pkNames.collect{|i| i + ' = :' + i}.join(' and ')};
END

    @deleteSQL = <<END
delete from #{@entityType.tableName}
where #{@entityType.pkNames.collect{|i| i + ' = :' + i}.join(' and ')};
END

    # find with PrimaryKey
    pkTypes = []
    @entityType.pkNames.each{|i| pkTypes << @entityType.fieldType[i]}
    addSelector('pk',pkTypes,<<END)
*
from #{@entityType.tableName}
where #{i = -1; @entityType.pkNames.collect{|name| i += 1; name + ' = :' + i.to_s}.join(' and ')};
END

    # select all
    addSelector('all',[],<<END)
*
from #{@entityType.tableName};
END
  end

  def insertEntity(p)
    record = @entityType.toRecord(p)

    statement = @insertSQL.dup
    @entityType.fieldType.keys.reverse.each{|i|
      statement.gsub!(/:#{i.to_s}/,"'#{record[i].to_s}'")
    }

    Transaction.getConnection.exec(statement)

    entity = @entityType.toEntity(p)

    changed
    notify_observers('insert',entity)

    return entity
  end

  def updateEntity(p)
    if @cached then
      oldEntity = find('pk',p)
    else
      oldEntity = nil
    end

    record = @entityType.toRecord(p)

    statement = @updateSQL.dup
    @entityType.fieldType.keys.reverse.each{|i|
      statement.gsub!(/:#{i.to_s}/,"'#{record[i].to_s}'")
    }

    Transaction.getConnection.exec(statement)

    entity = @entityType.toEntity(p)

    changed
    notify_observers('update',entity,oldEntity)

    return entity
  end

  def deleteEntity(p)
    if cached? then
      key = []
      @entityType.pkNames.each{|name| key << p[name]}
      oldEntity = find('pk',*key)
    else
      oldEntity = nil
    end

    record = @entityType.toRecord(p)

    statement = @deleteSQL.dup
    @entityType.fieldType.keys.reverse.each{|i|
      statement.gsub!(/:#{i.to_s}/,"'#{record[i].to_s}'")
    }

    Transaction.getConnection.exec(statement)

#    statement = Transaction.getConnection.createStatement(@deleteSQL)
#    @entityType.fieldType.keys.each{|i| statement.setParam(i,record[i])}
#    statement.execute

    changed
    notify_observers('delete',p,oldEntity)
  end

  def addSelector(selectorKey,parametorTypes,sql)
    @selector[selectorKey] = EntitySelector.new(@entityType,parametorTypes,sql)
    LogDev.add(SEV_INFO,"#{@entityType}Home.select( #{selectorKey} ) defined.",'EntitySelector')
  end

  def prepareCacher
    unless @cached then
      @cached = true
      addCacher('pk',@entityType.pkNames)
#      addCacherAll
    end
  end

  def addCacher(selectorKey,fieldNames)
    unless @cached then
      prepareCacher
    end

    begin
      Transaction.doTransaction {
	@selector[selectorKey] = EntityCacher.new(self,fieldNames)
	LogDev.add(SEV_INFO,"#{@entityType}Home.select( #{selectorKey} ) defined.",'EntityCacher')
      }
    rescue Exception
	LogDev.add(SEV_WARN,"#{@entityType}Home.select( #{selectorKey} ) failed to define.",'EntityCacher')
    end
  end

  def select(selectorKey,*arg)
    unless selector = @selector[selectorKey] then
      raise "No such selector [#{selectorKey}]"
    end

    return selector.select(*arg)
  end

  def find(selectorKey,*arg)
    unless selector = @selector[selectorKey] then
      raise "No such selector [#{selectorKey}]"
    end

    return selector.find(*arg)
  end

  # id等に使われる乱数を発生する
  def makeRandomInteger
    rand(IdBase) + IdBase
  end

  # プライマリキー:idを作成する
  # 'id' という integerのフィールドを持つテーブルで、且つそれが唯一のプライマリキーである場合にのみ使える
  def makeId
    while(true) # 無限ループだ! よくない。でも日常使うのには充分、と信じて使おう…。
      id = makeRandomInteger
      unless find('pk',id) then
	return id
      end
    end
  end


  # プライマリキー:id(連番)の用意をする
  # 'id' という integerのフィールドを持つテーブルで、且つそれが唯一のプライマリキーである場合にのみ使える
  def prepareSerialId
    @lastId = 0
    select('all').each{|e|
      id = e['id']
      if @lastId < id then
	@lastId = id
      end
    }
  end

  # プライマリキー:id(連番)を生成する
  # rollbackすると連番にならないがまぁいいでしょう.
  def makeSerialId
    return @lastId += 1
  end

  private

# cacher all にはバグアリ。
# 変更を見る、dupする、の2点で、実現は難しい。
#  def addCacherAll
#    selectorAll = [@selector['pk'].cache]
#
#    eval(<<END,binding)
#    class << selectorAll
#      def select(*arg)
#	return self[0].values.collect{|e| e[0]}
#      end
#      
#      def find(*arg)
#	return select()[0]
#      end
#    end
#END
#
#    @selector['all'] = selectorAll
#    LogDev.add(SEV_INFO,"#{@entityType}Home.select( all ) defined.",'EntityCacher')
#  end

  def cached?; return @cached; end
end


class Transaction
  include Observable

  attr('caches')
  attr('connection')

  ConnectionPool = Queue.new

  def initialize
    @mutexToMark = Mutex.new
    @connection = nil
    @caches = {}
    @toRollback = false

    beginTransaction
  end

  def markToRollback
    @mutexToMark.synchronize {
      @toRollback = true
      endTransaction
      beginTransaction
    }
  end

  def toRollback?
    return @toRollback
  end

  def beginTransaction
    unless @connection then
      @connection = ConnectionPool.pop
      @connection.exec('begin transaction;')
    end
  end

  def endTransaction
    if @connection then
      if @toRollback then
	rollback
      else
	commit
      end

      ConnectionPool.push(@connection)
      @connection = nil
    end
  end

  private

  def commit
    @connection.exec('commit;')
    
    changed
    notify_observers('commit',self)
  end

  def rollback
    @connection.exec('rollback;')
  
    changed
    notify_observers('rollback',self)
  end

  def Transaction.addConnection(connection)
    ConnectionPool.push(connection)
  end
  
  def Transaction.getConnection
    Transaction.current.connection
  end
  
  def Transaction.doTransaction
    unless Transaction.current then
      begin
	Transaction.continue(Transaction.new)
	yield(Transaction.getConnection)

      rescue Exception
	Transaction.current.markToRollback
	raise

      ensure
	Transaction.current.endTransaction
	Transaction.continue(nil)
      end
    end
  end

  def Transaction.markToRollback
    Transaction.current.markToRollback
  end

  def Transaction.toRollback?
    Transaction.current.toRollback?
  end
  
  def Transaction.continue(aTransaction)
    Thread.current['__transaction__'] = aTransaction
  end

  def Transaction.current
    Thread.current['__transaction__']
  end

  MutexToTrap = Mutex.new

  def Transaction.trapAndWrite(socket,content)
    MutexToTrap.synchronize {
      t = Transaction.current
      trap('SIGPIPE') {
	if t then
	  Transaction.continue(t)
	  Transaction.markToRollback
	end
	LogDev.add(SEV_WARN,
		   'SIGPIPE received. Client might be close this session. Continue...',
		   'ApplicationServer')
      }
      socket.write(content)
      trap('SIGPIPE','DEFAULT')
    }
  end
end



if $0 == __FILE__ #debug runner
  require 'postgres'
  require 'application'

  include Log::Severity
  LogDev = Log.new( STDERR )

  p 'testの準備'
  conn = PGconn.connect("localhost", 5432)
  Transaction.addConnection(conn)

  begin
    conn.exec('drop table test1;')
  rescue PGError
  end

  begin
    conn.exec(<<END)
create table test1 (
  id integer not null,
  name char(20),
  access_time integer,
  primary key (id)
);
END
  rescue PGError
  end

  begin
    # エンティティタイプ宣言
    Entity.define('Test1', # エンティティクラス Test1 の定義と、
		           # エンティティホーム Test1Home インスタンス生成
		  {
		    'id' => Integer, # フィールド名と型
		    'name' => String,
		    'access_time' => Time,
		  },
		  [
		    'id', # プライマリキーのリスト
		  ],
		  'test1' # テーブル名
		  )

    id = {}
    Test1Home.prepareCacher
    Test1Home.addSelector('name',[String],'* from test1 where name = :0;')

    p 'test1 レコードをinsertする'
    Transaction.doTransaction {
      Test1Home.prepareCacher
      (0 .. 3).each{|i|
	e = Test1Home.insertEntity({
				     'id' => Test1Home.makeId,
				     'name' => "entity A",
				     'access_time' => Time.now,
				   })
	Test1Home.updateEntity(e)
	id[i] = e['id']
      }
    }

    p 'test2 レコードをupdateする'
    Transaction.doTransaction {
      (1..2).each{|i|
	e = Test1Home.find('pk',id[i])
	e['name'] += " 1"
	Test1Home.updateEntity(e)
      }
    }

    p 'test3 レコードをdeleteするが、ロールバックさせる'
    Transaction.doTransaction {
      (2..3).each{|i|
	Test1Home.deleteEntity({'id' => id[i]})
      }

      Transaction.markToRollback
    }

    p 'test4 レコードをdeleteする'
    Transaction.doTransaction {
      (2..3).each{|i|
	Test1Home.deleteEntity({'id' => id[i]})
	id[i] = nil
      }
    }

    Transaction.doTransaction { Test1Home.select('all').each{|e| p e} }

  rescue PGError
    p $sql
    p $!
    p $@
  end
end

--------------E8723A67EE91239713117D17--