Skip to content
Permalink
Browse files

BatchInserter can now be used in migrations [#111] [#108]

  • Loading branch information...
andreas
andreas committed Feb 19, 2010
1 parent 170436b commit 2d54fb8bae6ac2df97d8af51f8a527194650f815
@@ -37,21 +37,19 @@ def []=(key, value)
#
class BatchInserter

# See class description for usage.
#
# See class description
#
def initialize(storage_path = Neo4j::Config[:storage_path]) # :yields: batch_inserter
# Neo4j must not be running while using batch inserter, stop it just in case ...
Neo4j::Transaction.finish
Neo4j.stop

# === Parameters
# storage_path:: optional, the location of the neo4j dabase on file system, default Neo4j::Config[:storage_path]
# db_version:: optional, sets version number on reference node, default nil -> do not set this property
def initialize(storage_path = Neo4j::Config[:storage_path], db_version=nil) # :yields: batch_inserter
# create the batch inserter
inserter = org.neo4j.kernel.impl.batchinsert.BatchInserterImpl.new(storage_path)

# save original methods
create_node_meth = Neo4j.method(:create_node)
create_rel_meth = Neo4j.method(:create_rel)
ref_node_meth = Neo4j.method(:ref_node)
ref_node_meth = Neo4j.method(:ref_node)

# replace methods
neo4j_meta = (class << Neo4j; self; end)
@@ -74,6 +72,7 @@ def initialize(storage_path = Neo4j::Config[:storage_path]) # :yields: batch_in

begin
yield inserter
Neo4j.ref_node[:db_version] = db_version if db_version
ensure
inserter.shutdown
# restore old methods
@@ -84,8 +83,22 @@ def initialize(storage_path = Neo4j::Config[:storage_path]) # :yields: batch_in
end
end
end
end

# This method is used if the batch inserter is used from the Migration API.
#
# === Parameters
# context:: the context on which the batch inserter code block is evaluated in, not used.
# version:: optional, if given then will set the property db_version on the context
def self.execute(context, version=nil, &block)
# Neo4j must not be running while using batch inserter, stop it just in case ...
Neo4j::Transaction.finish
Neo4j.stop

BatchInserter.new(Neo4j::Config[:storage_path], version, &block)

Neo4j.start
end
end

end

@@ -1,5 +1,6 @@
module Neo4j


# By including this mixing on a node class one can add migrations to it.
# Adds a db_version attribute on the class including this mixin.
#
@@ -12,12 +13,6 @@ def db_version
end


def init_with_node(java_node) # :nodoc:
super # call Neo4j::NodeMixin#init_with_node
migrate! self.class.migrate_to # for lazy migrations
end


# Force one or more migrations to occur if not already done yet.
# Will check the current db_version with the given 'to_version' and perform
# migrations. If the 'to_version' parameter is not given then it will upgrade the
@@ -38,13 +33,34 @@ def migrate!(to_version=nil, verbose = false)
# do we need to migrate ?
return if db_version == to_version

# ok, so we are running some migrations
# ok, so we are running some migrations
if (db_version < to_version)
Migrator.upgrade( (db_version+1).upto(to_version).collect { |ver| self.class.migrations[ver] }, self, verbose )
upgrade( (db_version+1).upto(to_version).collect { |ver| self.class.migrations[ver] }, verbose )
else
Migrator.downgrade( db_version.downto(to_version+1).collect { |ver| self.class.migrations[ver] }, self, verbose )
downgrade( db_version.downto(to_version+1).collect { |ver| self.class.migrations[ver] }, verbose )
end
end

# Running the up method on the given migrations.
#
# === Parameters
# migrations:: an enumerable of Migration objects
def upgrade(migrations, verbose=false)
migrations.each do |m|
puts "Running upgrade migration #{m.version} - #{m.name}" if verbose
m.up_migrator.execute(self, m.version, &m.up_block)
end
end

# Running the down method on the given migrations.
#
# === Parameters
# migrations:: an enumerable of Migration objects
def downgrade(migrations, verbose=false)
migrations.each do |m|
puts "Running downgrade migration #{m.version} - #{m.name}" if verbose
m.down_migrator.execute(self, m.version-1, &m.down_block)
end
self[:db_version] = to_version
end

def self.included(c) # :nodoc:
@@ -64,11 +80,14 @@ module ClassMethods
# up { ... }
# down { ... }
# end
#
# See the Neo4j::MigrationMixin::Migration which the DSL is evaluated in.
#
#
def migration(number, name, &block)
def migration(version, name, &block)
@migrations ||= {}
@migrations[number] = {:name => name, :block => block}
migration = Migration.new(version, name)
migration.instance_eval(&block)
@migrations[version] = migration
end

# This is used for lazy migration. It stores the version that we want to upgrade to but does not perform the migrations.
@@ -80,38 +99,59 @@ def migrate!(to_version=nil)
end
end

# This is used as both the context for the Migration DSL and running the actual migrations.
class Migrator # :nodoc:
attr_reader :up_blocks, :down_blocks
# This is the context in which the Migrations DSL are evaluated in.
class Migration < Struct.new(:version, :name)
attr_reader :up_block, :down_block, :up_migrator, :down_migrator

def up(&block)
@up_blocks ||= []
@up_blocks << block
# Specifies a code block which is run when the migration is upgraded.
#
# === Parameters
# migrator:: Default Neo4j::MigrationMixin::Migrator - used to execute the block
def up(migrator = Migrator, &block)
@up_block = block
@up_migrator = migrator
end

def down(&block)
@down_blocks ||= []
@down_blocks << block
# Specifies a code block which is run when the migration is upgraded.
#
# === Parameters
# migrator:: Default Neo4j::MigrationMixin::Migrator - used to execute the block
def down(migrator = Migrator, &block)
@down_block = block
@down_migrator = migrator
end

class << self
def upgrade(migrations, node_context, verbose)
get_blocks(migrations, verbose).up_blocks.each {|block| node_context.instance_eval &block}
end

def downgrade(migrations, node_context, verbose)
get_blocks(migrations, verbose).down_blocks.each { |block| node_context.instance_eval &block}
end
def to_s
"Migration version: #{version}, name: #{name}"
end
end

def get_blocks(migrations, verbose)
context = Migrator.new
migrations.each {|m| puts "Running Migration #{m[:name]}" if verbose; context.instance_eval &m[:block]}
context
# Responsible for running a migration
class Migrator
class << self
# Runs given migration block. If successful it will set the property
# ':db_version' on the given context.
#
# === Parameters
# context:: the context on which the block is evaluated in
# version:: optional, if given then will set the property db_version on the context
def execute(context, version=nil, &block)
context.instance_eval &block
context[:db_version] = version if version
end
end
end
end


# Overrides the init method so that it will check if any migration is needed.
# Migration might take place when the node is loaded.
#
module LazyMigrationMixin
def init_with_node(java_node) # :nodoc:
super # call Neo4j::NodeMixin#init_with_node
migrate! self.class.migrate_to # for lazy migrations
end
end
end

@@ -85,6 +85,8 @@ def start(neo_instance=nil)
Neo4j::Transaction.run do
Neo4j.event_handler.neo_started(self)
end

Neo4j::Transaction.run { @ref_node.migrate!}
nil
end

@@ -102,7 +102,8 @@ def run # :yield: transaction
tx = Neo4j::Transaction.new
ret = yield tx
rescue Exception => bang
#$NEO_LOGGER.warn{e.backtrace.join("\n")}
# puts "BANG #{bang}"
# puts bang.backtrace.join("\n")
tx.failure unless tx.nil?
raise
ensure
@@ -196,7 +197,7 @@ def success
# success() or failure() has been previously invoked.
#
def finish
raise NotInTransactionError.new unless Transaction.running?
return unless Transaction.running?
Neo4j.event_handler.tx_finished(self) unless failure?
begin
@neo_tx.success unless failure?
@@ -207,7 +208,6 @@ def finish
end

Thread.current[:transaction] = nil

if Lucene::Transaction.running?
$NEO_LOGGER.debug{"LUCENE TX running failure: #{failure?}"}

@@ -221,8 +221,6 @@ def finish

# Marks this transaction as failed, which means that it will inexplicably
# be rolled back upon invocation of finish().
#
# :api: public
def failure
raise NotInTransactionError.new unless Transaction.running?
@neo_tx.failure
@@ -61,20 +61,17 @@ class Foo
end

it "should be possible to use together with Migrations" do
pending "Endless recursion since it will trigger running the migration again"
Neo4j.migration 1, :first do
up do
puts "Create batch inserter" + caller.inspect
Neo4j::BatchInserter.new do
Neo4j.ref_node[:first] = true
end
Neo4j.start
up(Neo4j::BatchInserter) do
Neo4j.ref_node[:first] = true
end

down do
Neo4j.ref_node[:first] = nil
end
end

Neo4j.start
Neo4j.migrate!

Neo4j::Transaction.run do
@@ -29,16 +29,16 @@
end

it "should set the version on the ref node" do
Neo4j.stop

Neo4j.migration 1, :create_articles do
up do
end
down do
end
end

# when starting
Neo4j.db_version.should == 0
Neo4j.migrate!
Neo4j.start

# then
Neo4j.db_version.should == 1
@@ -165,6 +165,7 @@
class PersonInfo
include Neo4j::NodeMixin
include Neo4j::MigrationMixin
include Neo4j::LazyMigrationMixin
property :name
end

0 comments on commit 2d54fb8

Please sign in to comment.
You can’t perform that action at this time.