/* Ċ Ġ* ĠCopyright Ġ 2 0 1 7 Ġvr abel . z den ko @ gmail . com Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ package Ġnom ic . db Ċ Ċ import Ġnom ic . core .* Ċ import Ġnom ic . core . fact . Require Fact Ċ import Ġnom ic . core . script . InMemory Script Ċ import Ġnom ic . db . avro . Dependency Dto Ċ import Ġnom ic . db . avro . Installed Box Dto Ċ import Ġnom ic . hdfs . adapter . Hdfs Adapter Ċ import Ġorg . apache . avro . file . Data FileStream Ċ import Ġorg . apache . avro . file . Data FileWriter Ċ import Ġorg . apache . avro . specific . Specific Datum Reader Ċ import Ġorg . apache . avro . specific . Specific Datum Writer Ċ Ċ /** Ċ Ġ* Ġ@ author Ġvr abel . z den ko @ gmail . com Ċ Ġ*/ Ċ class ĠAv ro Db ( hdfs : ĠH dfs Adapter , Ġnom ic Home : ĠString ) Ġ: ĠN omic Db Ġ{ ĊĊ ĉ private Ġval Ġnom ic Home : ĠString ; Ċ ĉ private Ġval Ġh dfs : ĠH dfs Adapter ; ĊĊ ĉ init Ġ{ Ċĉ ĉ this . hdfs Ġ= Ġh dfs ; Ċĉ ĉ this . nom ic Home Ġ= Ġnom ic Home ; Ċ ĉ } ĊĊ ĉ override Ġfun Ġinsert OrUpdate ( box : ĠBox ): ĠBox Ġ{ Ċĉ ĉ // Ġconvert Ġto ĠD TO Ċĉ ĉ val Ġdto Ġ= Ġbox . to Dto () ĊĊĉ ĉ // Ġadd Ġadditional Ġdependencies Ġinto ĠD TO Ċĉ ĉ val Ġdependencies : List < Box Ref > Ġ= Ċĉĉ ĉ box . facts . filter Is Instance ( Require Fact :: class . java ). map Ġ{ Ġf Ġ-> Ġf . box Ġ} ĊĊĉ ĉ dependencies Ċĉĉ ĉ . map ( Box Ref :: to Dependency Dto ) Ċĉĉ ĉ . forEach Ġ{ Ġdep Ġ-> Ċĉĉĉ ĉ dto . get Dependencies (). add ( dep ) Ċĉĉ ĉ } ĊĊĉ ĉ // Ġsave Ġto Ġav ro Ġfile Ċĉ ĉ val Ġpath Ġ= Ġnormalized Path ( box . ref ()) Ċĉ ĉ hdfs . create ( path ). use Ġ{ Ċĉĉ ĉ val Ġbox Writer Ġ= ĠSpecific Datum Writer < Installed Box Dto >( Installed Box Dto :: class . java ) Ċĉĉ ĉ val Ġdata FileWriter Ġ= ĠData FileWriter < Installed Box Dto >( box Writer ) Ċĉĉ ĉ val Ġwriter Ġ= Ġdata FileWriter . create ( dto . schema , Ġit ) Ċĉĉ ĉ writer . append ( dto ) Ċĉĉ ĉ writer . close () Ċĉ ĉ } ĊĊĉ ĉ return Ġbox ; Ċ ĉ } ĊĊĊ ĉ override Ġfun Ġload All (): ĠList < Not Compiled Box > Ġ= Ċĉ ĉ hdfs . list Files ( nom ic Home , Ġtrue ) Ċĉĉ ĉ . filter ( this :: is Av ro File ) Ċĉĉ ĉ . flatMap ( this :: load Av ro File ) Ċĉĉ ĉ . map ( Installed Box Dto :: to Nom ic Box ) Ċĉĉ ĉ . toList () ĊĊĊ ĉ override Ġfun Ġload ( ref : ĠBox Ref ): ĠNot Compiled Box ? Ġ= Ċĉ ĉ load Av ro File ( normalized Path ( ref )) Ċĉĉ ĉ . map ( Installed Box Dto :: to Nom ic Box ) Ċĉĉ ĉ . first OrNull () ĊĊĊ ĉ override Ġfun Ġdelete ( ref : ĠBox Ref ) Ġ= Ċĉ ĉ hdfs . delete ( normalized Path ( ref )) ĊĊĊ ĉ override Ġfun Ġdepend ants Of ( ref : ĠBox Ref ): ĠList < Box Ref > Ġ= Ċĉ ĉ load All () Ċĉĉ ĉ . filter Ġ{ Ġbox Ġ-> Ċĉĉĉ ĉ box . dependencies . filter Ġ{ Ġdependency Ġ-> Ċĉĉĉĉ ĉ dependency . name Ġ== Ġref . name Ġ&& Ġdependency . group Ġ== Ġref . group Ċĉĉĉ ĉ }. isNotEmpty () Ċĉĉ ĉ } Ċĉĉ ĉ . map ( Box :: ref ) ĊĊĊĊĊ ĉ private Ġfun Ġis Av ro File ( file : ĠString ): ĠBoolean Ġ= Ċĉ ĉ file . endsWith (". avro ") ĊĊ ĉ private Ġfun Ġload Av ro File ( file : ĠString ): ĠSequence < Installed Box Dto > Ġ{ ĊĊĉ ĉ if Ġ(! hdfs . exist ( file )) Ġ{ Ċĉĉ ĉ return Ġempty Sequence () Ċĉ ĉ } ĊĊĉ ĉ hdfs . open ( file ). use Ġ{ Ċĉĉ ĉ val Ġbox Datum Reader Ġ= ĠSpecific Datum Reader < Installed Box Dto >( Installed Box Dto :: class . java ) Ċĉĉ ĉ val Ġdata Reader Ġ= ĠData FileStream < Installed Box Dto >( it , Ġbox Datum Reader ) Ċĉĉ ĉ val Ġlist Ġ= Ġdata Reader . toList () Ċĉĉ ĉ return Ġlist . as Sequence (); Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ private Ġfun Ġnormalized Path ( ref : ĠBox Ref ): ĠString Ġ{ Ċĉ ĉ val Ġnormalized Group Ġ= Ġref . group . replace ("/", Ġ"- "). replace (". "," - ") Ċĉ ĉ val Ġnormalized Name Ġ Ġ= Ġref . name . replace ("/", Ġ"- "). replace (".", Ġ"- ") Ċĉ ĉ return Ġ"${ nom ic Home }/${ normalized Group }.${ normalized Name }. avro " Ċ ĉ } Ċ Ċ } ĊĊ Ċ /** Ċ Ġ* Ġfunction Ġconvert Ġany Ġ[ Box ] Ġto ĠAv ro 's Ġ[ Installed Box Dto ] Ċ Ġ*/ Ċ private Ġfun ĠBox . to Dto (): ĠInstalled Box Dto Ġ= Ċ ĉ Installed Box Dto (). apply Ġ{ Ċĉ ĉ // Ġtransform Ġbasic Ġfields Ċĉ ĉ setName ( this @ to Dto . name ) Ċĉ ĉ set Group ( this @ to Dto . group ) Ċĉ ĉ set Version ( this @ to Dto . version ) ĊĊĉ ĉ // Ġtransform Ġscript Ċĉ ĉ this @ to Dto . script . open (). use Ġ{ Ċĉĉ ĉ set Script ( it . read Text ()) Ċĉ ĉ } ĊĊĉ ĉ // Ġtransform Ġdependencies Ċĉ ĉ val Ġdependencies Ġ= Ċĉĉ ĉ this @ to Dto . facts . as Sequence () Ċĉĉĉ ĉ . filter Is Instance ( Require Fact :: class . java ) Ċĉĉĉ ĉ . map ( Require Fact :: to Dto ) Ċĉĉĉ ĉ . toList () ĊĊĉ ĉ set Dependencies ( dependencies . to Mutable List ()) Ċ ĉ } ĊĊ Ċ /** Ċ Ġ* Ġfunction Ġconvert Ġ[ Require Fact ] Ġto ĠAv ro 's Ġ[ Dependency Dto ] Ċ Ġ*/ Ċ private Ġfun ĠRequire Fact . to Dto (): ĠDependency Dto Ġ= Ċ ĉ this . box . to Dependency Dto () Ċ Ċ private Ġfun ĠBox Ref . to Dependency Dto (): ĠDependency Dto Ġ= Ċ ĉ Dependency Dto (). apply Ġ{ Ċĉ ĉ setName ( this @ to Dependency Dto . name ) Ċĉ ĉ set Group ( this @ to Dependency Dto . group ) Ċĉ ĉ set Version ( this @ to Dependency Dto . version ) Ċ ĉ } Ċ Ċ private Ġfun ĠDependency Dto . to Nom ic Box Ref (): ĠBox Ref Ġ= Ċ ĉ Box Ref ( Ċĉ ĉ group Ġ= Ġthis !!. getGroup (), Ċĉ ĉ name Ġ Ġ= Ġthis . getName (), Ċĉ ĉ version Ġ= Ġthis . getVersion () Ċ ĉ ) Ċ Ċ private Ġfun ĠInstalled Box Dto . to Nom ic Box (): Not Compiled Box Ġ= Ċ ĉ Not Compiled Box ( Ċĉ ĉ name Ġ= Ġthis . getName () !! , Ċĉ ĉ group Ġ= Ġthis . getGroup () Ġ?: Ġ"", Ċĉ ĉ version Ġ= Ġthis . getVersion () Ġ?: Ġ"", Ċĉ ĉ script Ġ= ĠIn Memory Script ( this . get Script () !! ), Ċĉ ĉ dependencies Ġ= Ġthis . get Dependencies (). filter NotNull (). map ( Dependency Dto :: to Nom ic Box Ref ) Ċ ĉ ) Ċ