File Coverage

blib/lib/Data/Sync.pm

Criterion	Covered	Total	%
statement	125	539	23.1
branch	36	174	20.6
condition	3	36	8.3
subroutine	18	49	36.7
pod	14	40	35.0
total	196	838	23.3

#####################################################################
# Data::Sync
#
# Classes to make development of metadirectory/datapump apps
# simple and fast
#
# C Colbourn 2005
#
#####################################################################
# Revision
# ========
#
# 0.01	CColbourn		New module
#
# 0.02	CColbourn		Enhancements - see POD CHANGES
#
# 0.03	CColbourn		Enhancements - see POD CHANGES
#
# 0.04	CColbourn		Enhancements - see POD CHANGES
#
# 0.05	CColbourn		Bugfix - see POD CHANGES
#
# 0.06	CColbourn		Enhancements - see POD CHANGES
#
# 0.07	CColbourn		Enhancements - see POD CHANGES
#
# 0.08	CColbourn		Enhancements - see POD CHANGES
#
#####################################################################
# Notes
# =====
#
#####################################################################

use strict;
use warnings;


package Data::Sync;
our $VERSION="0.08";

#####################################################################
# New - constructor of datasync object
#
# takes parameters.
# returns blessed object
#
#####################################################################
sub new
{
	my $self = shift;
	my %synchash;
	my %params = @_;

	# make the object first!
	my $syncobject = bless \%synchash,$self;

	# define logging. If logging not set, use a coderef of return
	if ($params{'log'})
	{
		$syncobject->{'log'} = \&writelog;
		$syncobject->{'loghandle'} = $params{'log'};
	}
	else
	{
		$syncobject->{'log'} = sub{return}
	}
	if ($params{'configfile'})
	{
		my $return = $syncobject->load($params{'configfile'});
		if (!$return)
		{
			$self->{'log'}->($self->{'loghandle'},"ERROR: Not a readable config file ".$params{'configfile'});
			$self->{'lasterror'} = "Not a readable config file ".$params{'configfile'};
			return
		}
	}

	# assign the jobname (only needed if hashing or record mapping)
	if (!$params{'jobname'})
	{
		$params{'jobname'}="noname";
	}
	$syncobject->{'name'} = $params{'jobname'};

	# define the default multivalue record separator for concatenating LDAP multivalue attributes into a string
	$syncobject->{'mvseparator'} = "|";

	# put something to stdout for progress reporting. Convenience for debugging. Deliberately undocumented
	if ($params{'progressoutputs'})
	{
		if (!$params{'readprogress'}){$syncobject->{'readprogress'} = sub{print "R"}}
		if (!$params{'transformprogress'}){$syncobject->{'transformprogress'} = sub {print "T"}}
		if (!$params{'writeprogress'}){$syncobject->{'writeprogress'} = sub {print "W"}}
	}
	else
	{
		if(!$params{'readprogress'}){$syncobject->{'readprogress'} = sub {return}};
		if(!$params{'transformprogress'}){$syncobject->{'transformprogress'} = sub {return}};
		if(!$params{'writeprogress'}){$syncobject->{'writeprogress'} = sub {return}};
	}

	# return the object
	return $syncobject;

}

#####################################################################
# source
#
# defines the source data type & match critera
#
# takes $handle,\%search criteria
# returns true on successful definition
#####################################################################
sub source
{
	my $self = shift;
	my $handle = shift;
	my $criteriaref = shift;

	# do this regardless
	$self->{'readhandle'}=$handle;

	if (!$criteriaref)
	{
		# this /should/ mean the config is coming from a file
		# so return if the configs already been loaded (and
		# if not, it won't hurt anything to continue)
		if ($self->{'readcriteria'})
		{
			return 1;
		}
	}

	# assign the criteria hash as properties
	$self->{'readcriteria'} = $criteriaref;

	if (!$self->{'readcriteria'}->{'batchsize'})
	{
		$self->{'readcriteria'}->{'batchsize'}=0;
	}

	# Create coderef for LDAP
	if ($handle =~/LDAP/)
	{
		$self->{'read'} = \&readldap;
	}

	# everything else will be DBI/SQL
	else
	{
		$self->{'read'} =\&readdbi
	}

	1;
}

#########################################################
# readldap - read from an ldap datasource
#
# takes object as param
#
# returns result handle
#########################################################
sub readldap
{

	my $self = shift;
	my $result = $self->{'readhandle'}->search
		(filter=>$self->{'readcriteria'}->{'filter'},
		base=>$self->{'readcriteria'}->{'base'},
		scope=>$self->{'readcriteria'}->{'scope'},
		attrs=>$self->{'readcriteria'}->{'attrs'},
		controls=>[$self->{'readcriteria'}->{'controls'}]);
	if ($result->code)
	{
		$self->{'log'}->($self->{'loghandle'},"ERROR:".$result->error);
		return $result->error
	}
	else {return $result}
}

#########################################################
# readdbi - read from a dbi datasource
#
# takes object as param
#
# returns result handle
#########################################################
sub readdbi
{
	my $self = shift;

	my $stm = $self->{'readhandle'}->prepare($self->{'readcriteria'}->{'select'}) or return;
	my $result = $stm->execute;
	if ($result || $result eq "0E0"){return $stm}
	else
	{
		$self->{'log'}->($self->{'loghandle'},"ERROR: Could not read from database");
		$self->{'lasterror'}="ERROR: Could not read from database";
		return undef;
	}
}




#########################################################
# target - define the data target
#
# takes $handle, \%writecriteria
#
# returns 1 on success
#
#########################################################
sub target
{
	my $self = shift;
	my $handle = shift;

	$self->{'writehandle'} = $handle;

	my $criteriaref = shift;
	if (!$criteriaref)
	{
		# this /may/ mean the config is coming from a file
		# so return if the configs already been loaded (and
		# if not, it may be an ldap target so continue;
		if ($self->{'readcriteria'} && $handle!~/LDAP/)
		{
			return 1;
		}
	}
	else
	{
		$self->{'writecriteria'} = $criteriaref;
	}

	# LDAP index is /always/ DN, but an index is needed for hashing.
	if ($handle =~ /LDAP/){$criteriaref->{index} = "dn"}


	# Checking for fubars
	if ($criteriaref->{hashattributes} && !$criteriaref->{index})
	{
		$self->{log}->($self->{loghandle},"Can't set a target with hashing and no index!");
		$self->lasterror = "Can't set a target with hashing and no index!";
		return;
	}



	# create coderef to write to LDAP
	if ($handle =~/LDAP/)
	{
		$self->{'write'} = \&writeldap;
	}

	# write coderef for DBI
	if ($handle =~/DBI/)
	{
		$self->{'write'} = \&writedbi;
	}
	1;
}

########################################################
# writedbi
#
# takes object as param
#
# return t/f
########################################################
sub writedbi
{
	my $self = shift;
	my $writedata = shift;


	for my $line (@$writedata)
	{
		if ($line->{'Data::Sync::Action'} && $line->{'Data::Sync::Action'} eq "DELETE")
		{
			my $delete = "delete from ".$self->{writecriteria}->{table}. " where ";
			$delete .= $self->{writecriteria}->{index} ."=?";

			my $stm = $self->{writehandle}->prepare($delete);

			my $result = $stm->execute($line->{$self->{writecriteria}->{index}});
			if (!$result || $result eq "0E0")
			{
				$self->{'log'}->($self->{'loghandle'},"ERROR: Delete failed because ".$self->{'writehandle'}->errstr);
				$self->{'lasterror'}="ERROR: Delete failed because ".$self->{'writehandle'}->errstr;
			}

			next;
		}

		# Otherwise, the entry must be an update/add
		my @keys = keys %$line;
		my @values = map $_,values %$line;

		if ($self->{writecriteria}->{index})
		{
			my $update = "update ".$self->{'writecriteria'}->{'table'}. " set ";

			my @keys = keys %$line;
			my @values = map $_,values %$line;

			$update.=join "=?,",@keys;

			$update .="=? where ";
			$update .= $self->{'writecriteria'}->{'index'};
			$update .="=?";
			$self->{'log'}->($self->{'loghandle'},"Updating $update, ".join ",",@values);

			my $stm = $self->{'writehandle'}->prepare($update);

			my $result = $stm->execute(@values,$line->{$self->{'writecriteria'}->{'index'}});
			if (!$result || $result eq "0E0")
			{
				my $insert = "insert into ".$self->{'writecriteria'}->{'table'}." (";
				$insert .= join ",",@keys;
				$insert .=") VALUES (";
				$insert .=join ",",map { "?" } (0..scalar @values-1);
				$insert .=")";
				$self->{'log'}->($self->{'loghandle'},"Update failed, adding $insert, ".join ",",@values);
				$stm = $self->{'writehandle'}->prepare($insert);
				$result = $stm->execute(@values);
			}
			if (!$result || $result eq "0E0")
			{
				$self->{'log'}->($self->{'loghandle'},"ERROR: Add failed because ".$self->{'writehandle'}->errstr);
				$self->{'lasterror'}="ERROR: Add failed because ".$self->{'writehandle'}->errstr;
			}
			$self->{'writeprogress'}->($line->{$self->{'writecriteria'}->{'index'}});
		}
		else
		{
			my $insert = "insert into ".$self->{'writecriteria'}->{'table'}." (";
			$insert .= join ",",@keys;
			$insert .=") VALUES (";
			$insert .=join ",",map { "?" } (0..scalar @values-1);
			$insert .=")";
			$self->{'log'}->($self->{'loghandle'},"Adding $insert, ".join ",",@values);
			my $stm = $self->{'writehandle'}->prepare($insert);
			my $result = $stm->execute(@values);
		}
	}

}




########################################################
# writeldap - write to an ldap server
#
# takes object as param
#
# returns t/f
#########################################################
sub writeldap
{
	my $self = shift;
	my $writedata = shift;

	#if ($self->{'writecriteria'}->{'hashattributes'})
	#{
	#	my $checkedrecordset = $self->scanhashtable($writedata);
	#	$writedata = $checkedrecordset;
	#}


	foreach my $line (@$writedata)
	{
		my $dn = $line->{'dn'};

		delete $line->{'dn'}; # don't want the dn included in the hash of attrs to write

		if ($line->{'Data::Sync::Action'} && $line->{'Data::Sync::Action'} eq "DELETE")
		{
			my $result = $self->{'writehandle'}->delete($dn);
			if ($result->code)
			{
				$self->{'log'}->($self->{'loghandle'},"Delete of $dn failed because ".$result->error);
				$self->{lasterror} = "Delete of $dn failed because ".$result->error;
			}
			else
			{
				$self->{'log'}->($self->{'loghandle'},"Deleted $dn");
			}
			next;
		}

		# otherwise it's a modify or add


		$self->{'log'}->($self->{'loghandle'},"Modifying $dn, values ".join ",",values %$line);

		# experimental problem solution 20060212 - sunone/fedora and (possibly) AD will not permit a 'replace' on the objectclass
		# attribute in a modify - so we want to cover that off. Remove from the structure /unless/ a flag is set (document in ::Advanced)
		#my $modline = $line;
		#if (!$self->{writeobjectclass})
		#{
		#	for (keys %$modline)
		#	{
		#		print "$_ --\n";
		#		if (lc($_) eq "objectclass")
		#		{
		#			delete $$modline{$_};
		#		}
		#	}
		#}



		my $result =
			$self->{'writehandle'}->modify
			(
				dn=>$dn,
				replace=>[%$line]
			);


		if ($result->code)
		{
			$self->{'log'}->($self->{'loghandle'},"Modify failed '".$result->error."', adding $dn, values ".join ",",values %$line);
			$result =
				$self->{'writehandle'}->add
				(
					dn=>$dn,
					attrs=>[%$line]
				);

		}

		if ($result->code)
		{
			$self->{'log'}->($self->{'loghandle'},"ERROR: ".$result->error);
			$self->{'lasterror'}="ERROR: Add failed :".$result->error;

			return undef;
		}
		$self->{'writeprogress'}->("W");
	}
	return 1;
}




########################################################
# sourceToAoH
#
# Convert data from source to an array of hashes
# so that there's a standard form to write data out
#
# takes data handle (LDAP result or DBI)
#
# returns ref to AoH
#
########################################################
sub sourceToAoH
{
	my $self = shift;
	my $handle = shift;

	my @records;
	my $counter=1;

	# Convert LDAP
	if ($handle=~/LDAP/)
	{
		my $recordset = $self->ldapToAoH($handle);
		@records = @$recordset;
	}

	# convert DBI
	if ($handle=~/DBI/)
	{
		my $recordset = $self->dbiToAoH($handle);
		@records = @$recordset;
	}


	# if it's an empty recordset return unddef
	if (scalar @records == 0){return}

	# check against the hash records if defined and remove if the record has not changed.
	#if ($self->{'readcriteria'}->{'hashattributes'})
	#{
	#	my $checkedrecordset = $self->scanhashtable(\@records);
	#	@records = @$checkedrecordset;
	#
	#}

	return \@records;

}

#############################################################
# ldapToAoH - convert the content of an LDAP handle to an AoH
# takes - ldap handle
# returns - AoH
#############################################################
sub ldapToAoH
{
	my $self = shift;
	my $handle = shift;

	my @records;
	my $counter=1;


	if ($self->{'readcriteria'}->{'batchsize'} >0)
	{
		while ($counter<= $self->{'readcriteria'}->{'batchsize'})
		{
			my $entry=$handle->shift_entry;
			if (!$entry){last}
			my %record;
			for my $attrib ($entry->attributes)
			{
				$record{$attrib} = $entry->get_value($attrib);
			}
			$self->{'log'}->($self->{'loghandle'},"Read ".$entry->dn." from the directory");
			push @records,\%record;
			$counter++;
			$self->{'readprogress'}->($entry->dn);
		}
	}
	else
	{
		while (my $entry=$handle->shift_entry)
		{
			my %record;
			for my $attrib ($entry->attributes)
			{
				$record{$attrib} = $entry->get_value($attrib);
			}
			$self->{'log'}->($self->{'loghandle'},"Read ".$entry->dn." from the directory");
			push @records,\%record;
			$counter++;
			$self->{'readprogress'}->($entry->dn);
		}
	}





	return \@records;
}

#############################################################
# dbiToAoH - converts a db handle to an AoH
# takes - DB handle
# returns - AoH
#############################################################
sub dbiToAoH
{
	my $self = shift;
	my $handle = shift;

	my @records;
	my $counter=1;
	my $recordcounter=0;
	# this separation looks a bit strange, but combining into a single loop resulted in a segfault from DBI that I chased
	# for HOURS! resolve at a later date.
	if ($self->{'readcriteria'}->{'batchsize'} >0)
	{
		while ($counter <= $self->{'readcriteria'}->{'batchsize'})
		{
			my $entry = $handle->fetchrow_hashref;
			if (!$entry){last}
			my %record;
			for my $attrib (keys %$entry)
			{
				$record{$attrib} = $entry->{$attrib}
			}
			$self->{'log'}->($self->{'loghandle'},"Read entry ".++$recordcounter." from the database");
			push @records,\%record;
			$counter++;

			$self->{'readprogress'}->();
		}
	}
	else
	{
		while (my $entry = $handle->fetchrow_hashref)
		{
			my %record;
			for my $attrib (keys %$entry)
			{
				$record{$attrib} = $entry->{$attrib}
			}
			$self->{'log'}->($self->{'loghandle'},"Read entry ".++$recordcounter." from the database");
			push @records,\%record;
			$self->{'readprogress'}->();
		}
	}

	return \@records;
}

#############################################################
# hashrecord - take a record as a hash, and return the MD5
# hash
#
# takes hashref of record, arrayref of attribs to hash
#
#############################################################
sub hashrecord
{
	require Digest::MD5;
	my $self = shift;
	my $record = shift;
	my $attribs = shift;

	# make a hash of the current record
	my $attribstring;
	for (@$attribs)
	{
		if (!ref($_))
		{
			$attribstring .= $$record{$_};
		}
	}

	my $newhash = Digest::MD5->new;
	$newhash->add($attribstring);

	return $newhash->hexdigest;
}

#############################################################
# scanhashtable - run through a record set checking records
#  against the stored hash table
#  takes - a record set (AoH)
#  returns - a record set minus unchanged records.
#############################################################
sub scanhashtable
{
	my $self = shift;
	my $recordset = shift;

	require DBI;
	require Digest::MD5;

	my @records = @$recordset;
	my @hashcheckedrecords;

	my $hashdb = DBI->connect("DBI:SQLite:dbname=".$self->{'name'},"","") or die $!;

	# check the hash table for this database exists - if not, create it
	my $stm = $hashdb->prepare("select * from hashtable");

	if (!$stm)
	{
		$stm = $hashdb->prepare ("create table hashtable (targetkey CHAR(100),attribhash CHAR(32), status CHAR(1))");
		$stm->execute;
	}
	else
	{
		# tombstone previously deleted entries
		my $tmbstm = $hashdb->prepare("update hashtable set status=? where status=?");
		my $result = $tmbstm->execute("T","D");
		if (!$result || $result eq "0E0")
		{
			$self->{log}->($self->{loghandle},"Can't update status of previously deleted entries - expect write errors");
		}
		# mark all entries deleted, any that still exist will be marked update/exists
		my $delstm = $hashdb->prepare("update hashtable set status=? where status != ?");
		$result = $delstm->execute("D","T");
		if (!$result || $result eq "0E0")
		{
			$self->{log}->($self->{loghandle},"Can't update status of all entries - deltas may fail");
		}
	}

	my $getstm = $hashdb->prepare ("select attribhash from hashtable where targetkey=?");
	my $putstm = $hashdb->prepare("insert into hashtable (targetkey,attribhash) VALUES (?,?)");
	my $updstm = $hashdb->prepare("update hashtable set attribhash=? where targetkey=?");
	my $statusstm = $hashdb->prepare("update hashtable set status=? where targetkey=?");

	for my $record (@records)
	{
		$getstm->execute(${$record}{$self->{'writecriteria'}->{'index'}});
		my $oldhash = $getstm->fetchrow;

		my $newhash = $self->hashrecord($record,\@{$self->{writecriteria}->{hashattributes}});

		if (!$oldhash)
		{
			$putstm->execute(${$record}{$self->{'writecriteria'}->{'index'}},$newhash);
			push @hashcheckedrecords,$record;
			$statusstm->execute("N",${$record}{$self->{writecriteria}->{index}});
		}
		elsif($oldhash ne $newhash)
		{
			$updstm->execute($newhash,${$record}{$self->{'writecriteria'}->{'index'}});
			push @hashcheckedrecords,$record;
			$statusstm->execute("U",${$record}{$self->{writecriteria}->{index}});
		}
		else
		{
			$statusstm->execute("E",${$record}{$self->{writecriteria}->{index}});
		}
	}
	return \@hashcheckedrecords;
}

#############################################################
# getdeletes - get a list of the deleted records
# takes - null
# returns - arrayref of deleted entries
#############################################################
sub getdeletes
{
	my $self = shift;
	my $hashdb = DBI->connect("DBI:SQLite:dbname=".$self->{'name'},"","") or die $!;


	my $stm = $hashdb->prepare("select targetkey from hashtable where status=?");
	my $result = $stm->execute("D");


	if (!$result)
	{
		return
	}
	else
	{
		#return $stm->fetchall_arrayref();
		# fits into the existing code better to return a hash of index=>value
		my @deleteds;
		while (my $entry = $stm->fetchrow_array)
		{
			push @deleteds,{$self->{writecriteria}->{index} => $entry};
		}
		return \@deleteds;
	}
}

#############################################################
# deletes - define the behaviour for deleted records
#
# takes - hash of named params
# returns - success/fail
#############################################################
sub deletes
{
	my $self = shift;
	my $params;
	if (!@_){return} # don't set anything if blank
	if (scalar @_ == 1 && $_[0] =~/delete/i)
	{
		$params = "delete";
	}
	else
	{
		my %paramshash = @_;
		$params = \%paramshash;
	}

	$self->{deleteactions} = $params;

	return 1;
}


#############################################################
# Run - read the data, transform it, then write it.
#
# takes no parameters (apart from object)
# returns success or fail.
#
#############################################################
sub run
{
	my $self = shift;

	# fetch from source
	my $receivedata = $self->{'read'}->($self);

	# If we don't get anything back, return 0
	if (!$receivedata){return}

	my $result;

	my $AoHdata=[];
	while ($AoHdata)
	{
		# convert to an AoH
		my $AoHdata = $self->sourceToAoH($receivedata);
		if (!$AoHdata){last}

		# construct templated attributes
		$AoHdata = $self->makebuiltattributes($AoHdata);

		# remap attrib names to target names
		$AoHdata = $self->remap($AoHdata);

		# perform data transforms
		$AoHdata = $self->runtransform($AoHdata);

		if ($self->{validation})
		{
			my $result = $self->validate($AoHdata);
			if (!$result)
			{
				$self->{lasterror} = "ERROR: Validation failed";
				return undef;
			}

		}

		# check against hashtable
		if ($self->{writecriteria}->{hashattributes})
		{
			$AoHdata = $self->scanhashtable($AoHdata);
		}

		# handle deletions here - it MUST be after scanhashtable - note, it might be handy to put this in a separate function for ease of overloading.
		if ($self->{deleteactions})
		{
			my $deletes = $self->getdeletes();

			if ($deletes)
			{
				if ($self->{deleteactions} eq "delete")
				{
					for my $record (@$deletes)
					{
						# It's very unlikely that 'Data::Sync::Action' will collide with a true field name
						$record->{"Data::Sync::Action"} = "DELETE";
					}
				}
				else # set up an update
				{
					#$deletes = $self->runtransform($deletes,$self->{deleteactions});
					for my $entry (@$deletes)
					{
						%$entry = (%$entry,%{$self->{deleteactions}});
					}

				}
				# need to build attributes, remap and run transforms on deletes (for building DNs & field mappings)
				#$deletes = $self->makebuiltattributes($deletes);
				#$deletes = $self->remap($deletes);
				#$deletes = $self->runtransform($deletes);
				$AoHdata = \(@$AoHdata,@$deletes);
			}
		}

		# write to target
		$result = $self->{'write'}->($self,$AoHdata);

		# jump out if not in batch mode
		if ($self->{'readcriteria'}->{'batchsize'} == 0){last}

	}

	#set the timestamp
	my ($sec, $min, $hour, $mday, $mon, $year, $wday, $yday, $isdst) = localtime(time);
	$mon+=1;
	$year+=1900;
	$self->{'lastruntime'} = sprintf("%4d%02d%02d%02d%02d%02d",$year,$mon,$mday,$hour,$min,$sec);

	return $result;
}

#############################################################
# mappings - define mappings from source to target
#
# takes hash of sourceattrib=>targetattrib
#
# returns success or fail
#
#############################################################
sub mappings
{
	my $self = shift;
	my %params = @_;

	$self->{'map'} = \%params;

	return 1;
}

##############################################################
# remap - rename source keys in data to target keys
#
# takes data structure in AoH form
# returns data structure in AoH form
#
##############################################################
sub remap
{
	my $self = shift;
	my $data = shift;
	my @newdata;

	for my $line (@$data)
	{
		my %record;
		for my $attrib (keys %$line)
		{
			# retain unchanged name if nothing in map

			if ($self->{'map'}->{$attrib})
			{
				$record{$self->{'map'}->{$attrib}} = $$line{$attrib};
				$self->{'log'}->($self->{'loghandle'},"Remapped ".$attrib." to ".$self->{'map'}->{$attrib});
			}
			else
			{
				$record{$attrib}=$$line{$attrib}
			}
		}

		push @newdata,\%record;
	}

	return \@newdata;
}


##############################################################
# transforms - define transformations of data
#
# takes hash of params
# returns success or fail
#
##############################################################
sub transforms
{
	my $self = shift;

	$self->{transformations} = $self->maketrfunctions(@_);

	return 1;
}

##############################################################
# maketrfunctions - convert the various transform functions
#   into coderefs etc
#
# takes hash of params
# returns hash of coderefs etc
##############################################################
sub maketrfunctions
{
	my $self = shift;

	my %params=@_;

	# params are attrib=>regexstring or attrib=>coderef

	# if param is a regex, transform to a coderef
	for (keys %params)
	{
		# capture the concatenate special case
		if ($params{$_} =~/^concatenate$/){}
		# otherwise turn a function name into a coderef
		elsif ($params{$_} =~/^\w+$/)
		{
			$params{$_} = \&{$params{$_}};
		}
		elsif ($params{$_} !~/CODE/)
		{
			$params{$_}=eval "sub { #!#$_#!#
						my \$data=shift;
						\$data =~".$params{$_}.";
						return \$data;}";
		}
	}

	return \%params;


}

##############################################################
# runtransform - perform regexes and data transforms on
# the data
#
# takes AoH
# returns AoH
#
##############################################################
sub runtransform
{

	my $self = shift;
	my $inData = shift;
	my $transformations = shift;
	if (!$transformations){$transformations = $self->{transformations}}

	my @outData;

	for my $line (@$inData)
	{
		my %record;
		for my $attrib (keys %$line)
		{
			# only convert if there is a transform for this
			if ($transformations->{$attrib})
			{
				my $before = $$line{$attrib};
				# handle possible multi valued attribs
				$record{$attrib} = $self->recursiveTransform($$line{$attrib},$transformations->{$attrib});
			}
			else
			{
				$record{$attrib} = $$line{$attrib}
			}
		}
		push @outData,\%record;
		$self->{'transformprogress'}->();
	}

	return \@outData;

}

############################################################
# recursiveTransform - recursively transform data
#
# takes attrib,$transformation
#
# returns transformed attrib
#
############################################################
sub recursiveTransform
{
	my $self = shift;
	my $data = shift;
	my $transformation = shift;


	# if the transformation is to join values together
	if ($data =~/ARRAY/ && $transformation eq "concatenate")
	{
		my $string = join $self->{'mvseparator'},@$data;
		$data = $string;
	}
	# otherwise act on each instance
	elsif ($data =~/ARRAY/)
        {
                for (0..scalar @$data -1)
                {
                        $$data[$_] = $self->recursiveTransform($$data[$_],$transformation);
                }
        }

        elsif ($data =~/HASH/)
        {
                for my $inst (keys %$data)
                {
                        $$data{$inst} = $self->recursiveTransform($$data{$inst},$transformation);
                }
        }

        elsif ($transformation =~/CODE/)
        {
		my $before = $data;
		$data = $transformation->($data);
		$self->{'log'}->($self->{'loghandle'},"Transformed $before to ".$data);
	}

	return $data;
}

##########################################################
# buildattributes - fake up attributes from source data
#
# takes attribname=>'