From 1e6d50d33c6c38ce55ec89dda32f6b595320a029 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 13 Feb 2014 21:44:48 +0200 Subject: [PATCH] `Spec` improvements * Provide `id`-aware DSL methods for `MessageChannelSpec` * Rework `DslIntegrationConfigurationInitializer` to delegate to `IntegrationFlowBeanPostProcessor` to allow to use components from `IntegrationFlowBuilder` via `@Autowired` --- spring-integration-java-dsl/build.gradle | 7 +- .../gradle/wrapper/gradle-wrapper.jar | Bin 50557 -> 51106 bytes .../gradle/wrapper/gradle-wrapper.properties | 4 +- ...slIntegrationConfigurationInitializer.java | 88 +++------------ .../integration/dsl/EndpointSpec.java | 14 ++- .../integration/dsl/GenericEndpointSpec.java | 4 +- .../dsl/IntegrationFlowBeanPostProcessor.java | 106 ++++++++++++++++++ .../dsl/IntegrationFlowBuilder.java | 101 ++++++++--------- .../integration/dsl/IntegrationFlows.java | 19 +++- .../dsl/channel/DirectChannelSpec.java | 6 +- .../dsl/channel/ExecutorChannelSpec.java | 7 +- ...ort.java => LoadBalancingChannelSpec.java} | 2 +- ...ecSupport.java => MessageChannelSpec.java} | 20 ++-- .../dsl/channel/MessageChannels.java | 41 ++++++- .../dsl/channel/PriorityChannelSpec.java | 6 +- .../channel/PublishSubscribeChannelSpec.java | 2 +- .../dsl/channel/QueueChannelSpec.java | 15 ++- .../dsl/channel/RendezvousChannelSpec.java | 2 +- .../integration/dsl/core/Spec.java | 41 +++++++ .../integration/dsl/support/PollerSpec.java | 94 ++++++++++++++++ .../integration/dsl/support/Pollers.java | 70 ------------ .../dsl/test/IntegrationFlowTests.java | 53 ++++++--- 22 files changed, 443 insertions(+), 259 deletions(-) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java rename spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/{LoadBalancingChannelSpecSupport.java => LoadBalancingChannelSpec.java} (90%) rename spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/{ChannelSpecSupport.java => MessageChannelSpec.java} (85%) create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java create mode 100644 spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index de19646..2a5bb80 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -40,7 +40,7 @@ ext { eclipse.project.natures += 'org.springframework.ide.eclipse.core.springnature' -sourceSets.test.resources.srcDirs = ['src/test/resources', 'src/test/java'] +sourceSets.test.resources.srcDirs = ['src/test/java'] // See http://www.gradle.org/docs/current/userguide/dependency_management.html#sub:configurations // and http://www.gradle.org/docs/current/dsl/org.gradle.api.artifacts.ConfigurationContainer.html @@ -61,8 +61,7 @@ dependencies { } // enable all compiler warnings; individual projects may customize further -ext.xLintArg = '-Xlint:all,-options' -[compileJava, compileTestJava]*.options*.compilerArgs = [xLintArg] +[compileJava, compileTestJava]*.options*.compilerArgs = ['-Xlint:all,-options'] test { // suppress all console output during testing unless running `gradle -i` @@ -190,6 +189,6 @@ task dist(dependsOn: assemble) { task wrapper(type: Wrapper) { description = 'Generates gradlew[.bat] scripts' - gradleVersion = '1.10' + gradleVersion = '1.11' distributionUrl = "http://services.gradle.org/distributions/gradle-${gradleVersion}-all.zip" } diff --git a/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.jar b/spring-integration-java-dsl/gradle/wrapper/gradle-wrapper.jar index 5838598129719e795cc5633f411468bdec311016..3c7abdf12790879c06b07176de29647f77aa4129 100644 GIT binary patch delta 5953 zcmaJ_2UHVVv<)33G%-k*4xtO=(ZNCy>Afml={NptJ)VUA$b)16>0y_&z|_D zm3-;n9g`A~*OeZ&eA>!0o$M7ex4PzAMYylE!ZA9YtWw>Je(U8>$mn!0U_7(s@SG4P`|FW zTY4%%%(Z+qKBxDjZbW#pquX#9M~-6z1=b^At|?~hJFS{r^_B6b-;?vzb3~-c|H(ck z82>~sb5q_kZnBpnj^$(&O2HK?%32EF50?JGl};C{(d}PsGl~#p8B?p%YUsP<{cMq2 zuDb5yNNBgeK?c$zs(PI1DrO!KGcD|{$SC<%4jM*ECV zw-vthQWW^Y_4&wyWv7Bb{OKVOAza}C3AJ=UZZ&1aJ7xhYyjeMiCQk0g0FjAF@ zIYy5w{P1^(GDh$Y^I`!86lG9UksPCJhfP-Bx|Rdy-WYgM%!%B-ZOM6-=ZqIM>x6*8 z0yhZn6twqmx-v8kYeOG+Y-&j?S^VIEwvqvpDwy3z3cX@u|iVIAO-(6hUf zwvavjz41G0=GS!3yrJ5z^!2YAckggdmhif4gi zy5IMh)pNcy+PVmP9niq?GG)Qmyq^7K^6;2wX!aeoNg?!Qskn*an&7xQ%Km9LCQ^0gei5mFc`JS~W7u%+Kj#zq{3 z=AP;_-}P_GVqPJEZFW+bMKYq;U|+3_KJ%&+IvBUanR{W+i*k3+sB3nh85S`Rp30 zNpEld84-D@U4Ba+-I49waFC3S-s_<1pxg+sSWXSvEpH@F;GX9#Lb5X2&wqk zmM!Qdo$=01*Lxp6lru2zexApyvH|7byXoMPr=Dbr)DdT^2%$E^yf&oX{rMK|W5;w` zFkv=m?w;fNC__74SPt9hJaa7!_$oelwZFrJG?VWsf;p#b`*}5!n$v1vU@UBURquOU z1+QxAlgLK$Ceq>w`un=5&8$*e*|y^ERxCypZ|ab`4mCYYvRKQGIZDg4erutV>kpGt zPLQXLI80z0`gvBQ_|POT{&(K z?i*7;qH4U`jNU3LYgvYJu4c)aze@Uqe16<2hq_DT zVy=$7eo#i^^a6QX2TQ-n!NIAXirIM_hVGhAlI339j1+gVE=VZv)MK^io^^YZZJZ(* zcZpJd0Di~y$y*EzfeJTWIK6$dxv1VNx~m4Wo>F73d_S@MSx0;!Q)b0iFZ#&*`?eDX zn};$xqyE~|^6{2wi-Rm(rH-eBSuDe6emeCI!xQ^v*#50RI#g4$l>2@G`B+T?>~^_KB>^-TsEVU5}*0=gz4ZV#YeNN)y6jK z4DDCC*I7r`-%p!$UwHM(I%P8{U>7)3O!XBeE&L;o~m(u zuB7v>Npv?ewrL)9iv1T{byo^L-sI2kIf@8+Im%~bidj?-O$mz zeYeIl3Yo8jgGiRTR$`*A+?YJHeZsH3z4=tZB|% ztoc|dL&NC9IVYF9=R{V`VvIccZgo6$?l2QcRiXLCqFFN7X6>L_^1YC3+b%bKsnLmc z(JMvO_paGDFO+7RJLKdd~VS|8Z#oRk(}5%viwgQcO3_!l{5 z^VgYVR5(GnMahM=}l9tEtBAP(P+>mLd0t zb?``Uye(3kv6W6cROVfvesdj1~Nc1$h(~+X&K9!ojubxY>ml-8@^=YqDQc4;Xx`t=v*HrvIWFkj7CTaeY<$5to8o)5bZWA6)~ z1qSt~vU}5)KeOFO+T5q^+4U}X!7?cGo=M>Pat zq_OL3T8g{c*dH*YneigCOvHyuulpkdn%)~T^MYkVmW*PEM8p|tWro4bMb9AezP#*1 zCDLqkFI!Nrvo3tHkF>f&g)_Ixy1LF?8vcDF{HCmeeDqhDcNr}|A~w!@yD&`%DKGoa z(&YG~j6cjlD`(I`I?$j!?sNSvnw`6(lIfiV!sV@>9(KBRW>>7aWzv?X$bE%6+`<@} z#0{RYd{KZMPd8LvY0lbVUn#^Yv>qT=I*d8_r=)VztgUnNXvKbH=ary)WLKt!KeKvp zg_ZintT^X=(^i6aow@p<8yZaS-RJ))g)=h_RU`KfS#P`a#$h1DDXF3|PKQJB>^2hG zbGh?PRNu+%O`l(@$c`!NsLG+7ypd78MH`_&e&#SPvg(-N^(Qe{=+ruNl$&zZCw7^e zvT&p z3-B`MN2wikjC5ti86uD)GuVkR<-xZFe89S+7!-9d{4B ztLlC2(UbE~@?m<(K2ymRO9|FAy{#gP%C!V@^zG@oFFuGwiCq6r5n>UtOmH1l?WcZA zQza$l+3Zwsg-It#64lkK;go6pBx8|xHT$bwZdfU+Z2>01r`@KZbbC&uPHO8p3@0l$ zmG;*swB$Yw(8Y;ST`jic||3GW-a5YrR64sN0*)P#|h2S0(V0_ClAG-~i0q9FVV1phd)xEJa{MVR}A8X#f9dC)6v zX43(CUUCWuL<`6Z766)?1uH~um>x+%-7I1vm`J34CdHfN?|%_V0Zp<%6Iswi6xeKm zR=B2T5i}pvX!4hV9W*aAr3?lrw2D_4Wz~@rI2iIt$?$>Q{K}V{4ssMx0xg#$D)gSz z9QmVphB^`(UwG!oYM`Y5Xzp>T;SuiJt9xYBxB#BcNFjhkjK=>iuh%jN%fIEdhT+Bc zI{?FL*}q>70NvM$C|rF)zynl*KS0?>2P5R*Y&tmh4Y&&SVkD#|$o_tTz(Ian&>rC- z5S&HBk;PT~cZ2`ekQsy?8RINhaBZ8|e?7U~aJljr1df6yKY0E=FlY`SuksY&Q^|Y^ z7ekWmb|MQX$qx(wAJ7pHgFXatraU;z@+YgnB5-~JJo$bTfmaMVgd7;GW&z+;?06v2 zXAmNwBL$FO3Df$&zl!`<|HK7NQ3^^3#0ZR3mWYa)8b<|$R*B<%I=@8MI)DK%f?x?E z;AMI|&_Rj{_)v8U4>peCa$f`^WdoN)ng~q8ga^x2!wDfd@rx6iz7=l19V6T=CDQ^2U0s#i#NnApJ1t3v(ng@3i;dfpTa2(ug zSc%vKEP-ovLYHtL-2CsdXZQoc6)(RP;YK+sduK0qqSD71-`e=%unbz?12v!iV~k6L zi4{Onua1`?WAB>u6#Q`oJVl5=w`_s@V1D4hnhLm6555nKK$M7l)S?0Cbrs+`_9#+xGtNh1G+t(NGoe0gVoNMMm>`h1;HnTyRE`Ch zS_nQYjocHTfa!pPVu)|{)ifMiN;?IR(87k-L-Y(G(lea_VsUH2iUiv0Pjoh3Co{s3RIiy ztAL5J1``D*qG>P0FUGMK41be#4~wI@P=rrbKbBB$07N+*m+O5{jUo|6d?_$pqWss~ zu?=T#{|YE8mBT3@ta1x}w!EB>Ga-C*xTl;J$c4KJPZ4oyRRRw&yevk#xP2D4A4)^8 LU?p@_;{N*|!W@tg delta 5305 zcmZu#2{=@38$R}3wi)|Uc8!dkB$X}6n#S0(mVHSJ$uyCzg`6xEBfAtvd}H6&&rbCz zvNfTyQz6lRX2!4Icm3ybuJd@_`+4sBdEWQD&MPaPW;>3C-Q1XlmKgvT8G)*E4|N~1 zOEQt3Dv7)wv>zTr@z8#qo~8!=1UgHAyzg*r~`LsGFHy&5<_m@#vt*$=9xGZ*DGWteW)w&;S{i8PX z_>0M#;k8c1m=kSh!pxhDVBP^4abkI;MYXVnr-g6vOP>Fw)CLu#DzxIgyrpSeQW*QI0TGCU-o#e`|#W$1J8|P;&fMhkh2h4gx|>@ywA2s1zl(=z6iZ7rM-X7fXUmyh%H(Lh8`gOqSmZ} z7GA@ns9Y-zSFfzOKB83R*>Nw&p3@xhx?IaONB-$QMLA6=->ZTQh@}D1ni$nw>Gs@L z+w5%Eu56EwjMz_D-#ZNS@zTDog*x4G49{G5_!N0Ak31N>D}oR{PSl(|?U;nPl^bin zZFvf#y*BOb&yDMuLDb%J zYrSbYpvXA~wPG1>=1xK8^m2@fxXF?D!k>qfPQHA*=+caDX`va5=-?)VBMu`ye+9N@ zhxNCgin1D!Y4K>qC$j7J9*h4KQ+VtnS0&ce0wM01_p)FRCviX1j0w@*^-p@L>H9^c zc&}*7(v6K1EM(!{E=<}og)&djaZk~N z;=CJ0Y}BwSL%kDL%5D9*lX6)G+}*7~`Y#9ciWucwa-~qQYdod4+}@a0+mSg|O@ng^ zk?kksoICRF&F4KjmJ@U330gt_yKr9A6AdTx5eMPWL(U_aff326S-*7$D{*H{Ul(My z}4sU$I1`&wCR- zeCt*=i@20yeWN!)m#Zrw5;LRx^iU2L@slO-yp!sngqnAjO~Ry$i`!RSnJJza^PrKK zcJ_{YE9QtmfnRX_NYy~9EW~pWNi1#-dG1(>e`d&2w~fC}yDM0=!cBZ&yozhBYWnss zV;3wp&w3C#dwd=TfAu@>d}Fv(^0pX5fU1JP9j1Ca<|Ew>A@Eb&T6&nnEM}E$bi;a2 z-Ol&sD8;i&7aJp^f6myo(ZgR~{+#XZVL)8KjqR{<++2IR#;W>KfyPnsx%6mJ?3+gq z@%>V^aBIRwOKI#Q{ZU^;;Xqqgr-AiX`$K)_1^bU=T<&?Ej0oe7N*`tMp53wvoUm3J z=AxgeT7Q%#ktpI!=SUkYzmm=k=4{Fm@vu_QAj?ZR{2WW<>g`R*DKGm^4sO;JDG8lWm6v{@Yn23o*jk)Ai9hdL# zbN5%BcXWV@_bTNPt82^9Mt=4|e)X%>IC&fNT!MUSQkv4}y^9_}`Cr0KUuN__kZe|zd#Smu#=u96?YH(na)#ku+Czuzhv^$_=6290#9pY>E8&+YImasS?S<=K=R7t$g>IP_w4 z^fmVDX)8hfO7Uf;9vaThX4Q)VGS!yq*VDCnRYzyXKX}b)uS-_M-QfyNbIQu^=;N&) zp^3-Ew42y`66_Z$-+Ul5Z>j$_wa+=_$r2jpk!^N0G%j3i-p6bd6YTfMXH^9~qr>7J zrpjfwk)@@Iezuu5pHg+Q(+oBmoPuX}E~k>wHy_v9fVV1d@?3{c@Mfm1MQC2$EXnc? zn%Pd9(_Ev%ajbSn5jCuv!vNCVZPY&DMR?|Kei@QI|u~5Z+ zbnDpWO13<)Tn57lU$HT?>ve8HyIvr&s`X|)$m>F=P9x+C8d5(33k7)x)iPTw+^bmb z@Fu9HRd^c0&WA61@cDJ?8PPddE$!eCF-h8*rIu4@n#f8;mO)&m-4tV*0nKj#8WU`< zr%L~`Tn;}c_78CgA)R*RfNMh+EEjLSWy}veDZraoDv7yYq`m%3Nh#Xb)kL{+gD1r2 zfPG^7gHemayfj7i(F1CzAq|bY>B^PgtPD~elq=_RT3*Bvs$|6f7FY$83z#G9>$p^V zaBqX{X`=Y)vu)bs|!}pLhm`Um28Z z87v0i8r{n$kJm3<4!6Ha7{FbSm$^vy+)W$ZwCvbzi^o#E)iY|z8EHcLxW+BHURp<1 z=03RpX+Y1!f5yfpaAYN@_V%22)z33g#bWMs^2>4^7{iW0V}m8T5YwLp`HJSa_MK64 zBK=|MI?r!LPZ@q*44g)PvbbbAaMYNljIYNjQDNP+GmvXF?Xq@TDL%2P9BcFXtTW-< zt?tKKESY0OHv#k+!76FvOwp1)cw-Se8{Uh(cCvoX+9$p6jH?+se)H} zRj#CJ6>yc!0;z;lnE~mnWts9Jq_^--hb~avS{$PqU*p$;WYxd6eTGvv-Zq&*evFyN z#x_{~+cPe@WVe`CgZ(9GQud4@tK%#q@<(=RC0WkTCSu>wgtgJWJ8P?&e-_y%b@Xyu zJMYNd%0x?_B}bMojdyKM7r#-7YHTl6*S&LbNO9$8G|**hR7_{5#`JAW*r>JN2UVM# z7h%I;)7d*eBPiFNs4AF;-7z<1#vSN2>>2eJ#oajGd9r{Tl#Z;{mGqWZ4=`~mTwzK0 zD)Kv_Nc4m2Q)S-H`{4Eju~+PykU_>|rpAEipH`7d1QpSTX-P)e1+)?k*#eX8KC_ct z6Cd>rbEyGfogTbUC=Kq^3zqfSmrzsSSf|gsmaj`96~#y6CdkN1iX`%XKQ#nJj?x2w zI>_NB0dnNgm)ZEt?}ixpAEj`WMF$PhQGy+=Z|{|Wl^t<`qXHoBROyn-_j=nCy?B+uio}i^+{8%?cy3=CD1UE z3b`gT-3g)zdi&Rkip8I`QcgHXKCglA3Ce#?k|3-?(}J|vprarFHG+K6?mhv416Vkj z3jO9EbWy?Fn2MT*=Ff?g1p03T%f5(OFrz|%LRMG7N4s!xmU0GMtYD>hcOY*LR)C6p zGDK1{ff|+Elo7K^(#Q^FoqKu1QTS0N~Iy<{(e z1@a1k$?V)9Up1>3salJh!k*Jmnfy>W$UW@H~`4(11cy$Ks7RG|B3q#DhFGsNzTSxD^BT&VlAmt3wmVl z7lZ%PODq{-XiP@|l8FDEN76DTZ)6aViwXds{~hdIIar(X;0bF1u&$1k{5F8Sb)#~|l=8`k5Kjah0BG;ydG0_4LOTfP*Kmjo{w&V6Y6#V0 z3l|0F$UcMfdV9keQ=g|Ow#=bOn zuI};v-9%};xJs0{9K=`2@$c8;_I2=5s0?_dc~|=JW)j~r2DEKHO|JUvi$;DUNG>hZ zOM`tptynToZ-nGtZMIP_QC^$?&;w1{{?{-t5&|Cr(Jfq*dGl)_@rWjqQxvs`P`v$Q zFVEw>0{j~2L4nrY7-TCcX8kF&RQSOk=@NU(1x&An)j}G@AdQH98mTg&@(zLj+qlTJ zLk}NA;6nO;Pa{LlF0as^{yVmjIJfe^se;3Md@2%b6JMdehe3Plv3-2u1>j)Y8A>st z?WEA=m!MX=A-R?3sNrvDAm$2aOHtYvmQ+k;#^E_h%%Y_EWe7i7p#^btL3@wZKJaHS qvt0s|?2sZSCH++)&5Is%i|)gPm4n_Lvg`_g0&oeM_uu8D(f integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.class, false, false); + private void populateBeansFromSpecs(ConfigurableListableBeanFactory beanFactory) { + Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory, + "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + + "Consider using 'GenericApplicationContext' implementation." + ); + Map specs = beanFactory.getBeansOfType(Spec.class, false, false); BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory; - for (Map.Entry integrationFlowEntry : integrationFlows.entrySet()) { - String flowName = integrationFlowEntry.getKey(); - String flowNamePrefix = flowName + ":"; - IntegrationFlow flow = integrationFlowEntry.getValue(); - int channelNameIndex = 0; - for (AbstractBeanDefinition beanDefinition : flow.getIntegrationComponents()) { - if (beanDefinition instanceof InstanceBeanDefinition) { - final Object instance = beanDefinition.getSource(); - Collection values = beanFactory.getBeansOfType(instance.getClass(), false, false).values(); - if (!values.contains(instance)) { - if (instance instanceof AbstractMessageChannel) { - String channelBeanName = ((AbstractMessageChannel) instance).getComponentName(); - if (channelBeanName == null) { - channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; - } - registry.registerBeanDefinition(channelBeanName, beanDefinition); - } - else if (instance instanceof EndpointSpec) { - EndpointSpec endpointSpec = (EndpointSpec) instance; - MessageHandler messageHandler = endpointSpec.getHandler(); - ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint(); - String id = endpointSpec.getId(); - - String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); - String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; - BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); - BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); - - String endpointBeanName = id; - if (endpointBeanName == null) { - endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint); - } - registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint)); - } - else { - String beanName = generateInstanceBeanDefinitionName(registry, instance); - registry.registerBeanDefinition(beanName, beanDefinition); - } - } - } - else { - BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); - } - } - registry.removeBeanDefinition(flowName); - beanFactory.destroyBean(flowName); + for (Map.Entry specEntry : specs.entrySet()) { + String id = specEntry.getKey(); + Spec spec = specEntry.getValue(); + registry.removeBeanDefinition(id); + beanFactory.destroyBean(id); + beanFactory.registerSingleton(id, spec.get()); } - - } - - @SuppressWarnings("serial") - private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) { - return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() { - - @Override - public String getBeanClassName() { - return instance.getClass().getName(); - } - }, registry); } } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java index 3f05de0..cfce04d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EndpointSpec.java @@ -23,6 +23,7 @@ import java.util.List; import org.aopalliance.aop.Advice; import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageHandler; @@ -31,17 +32,17 @@ import org.springframework.messaging.MessageHandler; * @author Artem Bilan * @since 4.0 */ -public abstract class EndpointSpec, C extends MessageHandler> { +public abstract class EndpointSpec, H extends MessageHandler> { private final ConsumerEndpointFactoryBean endpointFactoryBean = new ConsumerEndpointFactoryBean(); - private final C messageHandler; + private final H messageHandler; private final List adviceChain = new LinkedList(); private String id; - EndpointSpec(C messageHandler) { + EndpointSpec(H messageHandler) { this.messageHandler = messageHandler; this.endpointFactoryBean.setHandler(this.messageHandler); if (this.messageHandler instanceof AbstractReplyProducingMessageHandler) { @@ -78,6 +79,11 @@ public abstract class EndpointSpec, C extends Messa return _this(); } + public S poller(PollerSpec pollerMetadataSpec) { + this.endpointFactoryBean.setPollerMetadata(pollerMetadataSpec.get()); + return _this(); + } + String getId() { return id; } @@ -86,7 +92,7 @@ public abstract class EndpointSpec, C extends Messa return this.endpointFactoryBean; } - C getHandler() { + H getHandler() { return this.messageHandler; } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java index 77211a6..8590de3 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/GenericEndpointSpec.java @@ -22,9 +22,9 @@ import org.springframework.messaging.MessageHandler; * @author Artem Bilan * @since 4.0 */ -public final class GenericEndpointSpec extends EndpointSpec, C> { +public final class GenericEndpointSpec extends EndpointSpec, H> { - GenericEndpointSpec(C messageHandler) { + GenericEndpointSpec(H messageHandler) { super(messageHandler); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java new file mode 100644 index 0000000..30d448a --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java @@ -0,0 +1,106 @@ +package org.springframework.integration.dsl; + +import java.util.Collection; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactoryUtils; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.parsing.BeanComponentDefinition; +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; +import org.springframework.beans.factory.support.BeanDefinitionRegistry; +import org.springframework.beans.factory.support.GenericBeanDefinition; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.config.ConsumerEndpointFactoryBean; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.dsl.config.InstanceBeanDefinition; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor { + + private final ConfigurableListableBeanFactory beanFactory; + + private final BeanDefinitionRegistry registry; + + + public IntegrationFlowBeanPostProcessor(ConfigurableListableBeanFactory beanFactory) { + Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory, + "To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." + + "Consider using 'GenericApplicationContext' implementation." + ); + this.beanFactory = beanFactory; + this.registry = (BeanDefinitionRegistry) beanFactory; + } + + @Override + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof IntegrationFlow) { + String flowNamePrefix = beanName + ":"; + int channelNameIndex = 0; + for (AbstractBeanDefinition beanDefinition : ((IntegrationFlow) bean).getIntegrationComponents()) { + if (beanDefinition instanceof InstanceBeanDefinition) { + final Object instance = beanDefinition.getSource(); + Collection values = this.beanFactory.getBeansOfType(instance.getClass(), false, false).values(); + if (!values.contains(instance)) { + if (instance instanceof AbstractMessageChannel) { + String channelBeanName = ((AbstractMessageChannel) instance).getComponentName(); + if (channelBeanName == null) { + channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++; + } + registry.registerBeanDefinition(channelBeanName, beanDefinition); + } + else if (instance instanceof EndpointSpec) { + EndpointSpec endpointSpec = (EndpointSpec) instance; + MessageHandler messageHandler = endpointSpec.getHandler(); + ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint(); + String id = endpointSpec.getId(); + + String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler); + String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null; + BeanComponentDefinition definitionHolder = new BeanComponentDefinition( + new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias); + BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry); + + String endpointBeanName = id; + if (endpointBeanName == null) { + endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint); + } + registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint)); + } + else { + String name = generateInstanceBeanDefinitionName(registry, instance); + registry.registerBeanDefinition(name, beanDefinition); + } + } + } + else { + BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry); + } + } + } + return bean; + } + + @Override + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + @SuppressWarnings("serial") + private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) { + return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() { + + @Override + public String getBeanClassName() { + return instance.getClass().getName(); + } + }, registry); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index d113b1c..cbd6483 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -23,6 +23,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory import org.springframework.integration.core.GenericSelector; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessageSelector; +import org.springframework.integration.dsl.channel.MessageChannelSpec; import org.springframework.integration.dsl.support.EndpointConfigurer; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.integration.filter.MessageFilter; @@ -34,6 +35,7 @@ import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.integration.transformer.MethodInvokingTransformer; import org.springframework.integration.transformer.Transformer; import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; /** * @author Artem Bilan @@ -62,10 +64,53 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder channel(MessageChannel messageChannel) { + Assert.notNull(messageChannel); this.currentMessageChannel = messageChannel; return this.addComponent(this.currentMessageChannel).registerOutputChannelIfCan(this.currentMessageChannel); } + public IntegrationFlowBuilder channel(MessageChannelSpec messageChannelSpec) { + Assert.notNull(messageChannelSpec); + return this.channel(messageChannelSpec.get()); + } + + public IntegrationFlowBuilder transform(String expression) { + return this.transform(PARSER.parseExpression(expression)); + } + + public IntegrationFlowBuilder transform(Expression expression) { + return this.transform(new ExpressionEvaluatingTransformer(expression)); + } + + public IntegrationFlowBuilder transform(GenericTransformer genericTransformer) { + return this.transform(genericTransformer, null); + } + + public IntegrationFlowBuilder transform(GenericTransformer genericTransformer, + EndpointConfigurer> endpointConfigurer) { + Transformer transformer = genericTransformer instanceof Transformer + ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); + return this.register(new GenericEndpointSpec(new MessageTransformingHandler(transformer)), endpointConfigurer); + } + + public IntegrationFlowBuilder filter(String expression) { + return this.filter(PARSER.parseExpression(expression)); + } + + public IntegrationFlowBuilder filter(Expression expression) { + return this.filter(new ExpressionEvaluatingSelector(expression)); + } + + public IntegrationFlowBuilder filter(GenericSelector genericSelector) { + return this.filter(genericSelector, null); + } + + public IntegrationFlowBuilder filter(GenericSelector genericSelector, EndpointConfigurer endpointConfigurer) { + MessageSelector selector = genericSelector instanceof MessageSelector + ? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector); + return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer); + } + private IntegrationFlowBuilder registerOutputChannelIfCan(MessageChannel outputChannel) { this.flow.addComponent(outputChannel); if (this.currentComponent != null) { @@ -83,50 +128,10 @@ public final class IntegrationFlowBuilder { return this; } - public IntegrationFlowBuilder transform(String expression) { - return this.transform(PARSER.parseExpression(expression)); - } - - public IntegrationFlowBuilder transform(Expression expression) { - return this.transform(new ExpressionEvaluatingTransformer(expression)); - } - - public IntegrationFlowBuilder transform(GenericTransformer genericTransformer) { - Transformer transformer = genericTransformer instanceof Transformer - ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); - return this.transform(genericTransformer, new DefaultEndpointConfigurer>()); - } - - public IntegrationFlowBuilder transform(GenericTransformer genericTransformer, - EndpointConfigurer> endpointConfigurer) { - Transformer transformer = genericTransformer instanceof Transformer - ? (Transformer) genericTransformer : new MethodInvokingTransformer(genericTransformer); - GenericEndpointSpec spec = new GenericEndpointSpec(new MessageTransformingHandler(transformer)); - endpointConfigurer.configure(spec); - return this.register(spec); - } - - public IntegrationFlowBuilder filter(String expression) { - return this.filter(PARSER.parseExpression(expression)); - } - - public IntegrationFlowBuilder filter(Expression expression) { - return this.filter(new ExpressionEvaluatingSelector(expression)); - } - - public IntegrationFlowBuilder filter(GenericSelector genericSelector) { - return this.filter(genericSelector, new DefaultEndpointConfigurer()); - } - - public IntegrationFlowBuilder filter(GenericSelector genericSelector, EndpointConfigurer endpointConfigurer) { - MessageSelector selector = genericSelector instanceof MessageSelector - ? (MessageSelector) genericSelector : new MethodInvokingSelector(genericSelector); - FilterEndpointSpec spec = new FilterEndpointSpec(new MessageFilter(selector)); - endpointConfigurer.configure(spec); - return this.register(spec); - } - - private IntegrationFlowBuilder register(EndpointSpec endpointSpec) { + private > IntegrationFlowBuilder register(S endpointSpec, EndpointConfigurer endpointConfigurer) { + if (endpointConfigurer != null) { + endpointConfigurer.configure(endpointSpec); + } MessageChannel inputChannel = this.currentMessageChannel; this.currentMessageChannel = null; if (inputChannel == null) { @@ -143,12 +148,4 @@ public final class IntegrationFlowBuilder { return this.flow; } - private class DefaultEndpointConfigurer> implements EndpointConfigurer { - - @Override - public void configure(S spec) { - - } - } - } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index ba60dd3..a85a47f 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -18,9 +18,11 @@ package org.springframework.integration.dsl; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.dsl.channel.MessageChannelSpec; +import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; /** * @author Artem Bilan @@ -31,8 +33,17 @@ public final class IntegrationFlows { return new IntegrationFlowBuilder().channel(messageChannel); } + public static IntegrationFlowBuilder from(MessageChannelSpec messageChannelSpec) { + return from(messageChannelSpec.get()); + } + public static IntegrationFlowBuilder from(MessageSource messageSource) { - return from(messageSource, null); + return from(messageSource, (PollerMetadata) null); + } + + public static IntegrationFlowBuilder from(MessageSource messageSource, PollerSpec pollerSpec) { + Assert.notNull(pollerSpec); + return from(messageSource, pollerSpec.get()); } public static IntegrationFlowBuilder from(MessageSource messageSource, PollerMetadata pollerMetadata) { @@ -45,9 +56,9 @@ public final class IntegrationFlows { .currentComponent(factoryBean); } - public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) { + /*public static IntegrationFlowBuilder from(AbstractEndpoint endpoint) { return new IntegrationFlowBuilder(); - } + }*/ private IntegrationFlows() { } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java index 4d06754..15dc281 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/DirectChannelSpec.java @@ -21,10 +21,10 @@ import org.springframework.integration.channel.DirectChannel; /** * @author Artem Bilan */ -public class DirectChannelSpec extends LoadBalancingChannelSpecSupport { +public class DirectChannelSpec extends LoadBalancingChannelSpec { @Override - public DirectChannel get() { + protected DirectChannel doGet() { this.channel = new DirectChannel(this.loadBalancingStrategy); if (this.failover != null) { this.channel.setFailover(this.failover); @@ -32,7 +32,7 @@ public class DirectChannelSpec extends LoadBalancingChannelSpecSupport { +public class ExecutorChannelSpec extends LoadBalancingChannelSpec { private final Executor executor; @@ -31,7 +31,8 @@ public class ExecutorChannelSpec extends LoadBalancingChannelSpecSupport, C extends AbstractMessageChannel> extends ChannelSpecSupport { +public abstract class LoadBalancingChannelSpec, C extends AbstractMessageChannel> extends MessageChannelSpec { protected LoadBalancingStrategy loadBalancingStrategy = new RoundRobinLoadBalancingStrategy(); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java similarity index 85% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java index 8ea74b6..74951e8 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/ChannelSpecSupport.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java @@ -19,12 +19,13 @@ package org.springframework.integration.dsl.channel; import java.util.Arrays; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.dsl.core.Spec; import org.springframework.messaging.support.ChannelInterceptor; /** * @author Artem Bilan */ -public abstract class ChannelSpecSupport, C extends AbstractMessageChannel> { +public abstract class MessageChannelSpec, C extends AbstractMessageChannel> extends Spec { protected C channel; @@ -34,6 +35,11 @@ public abstract class ChannelSpecSupport, C e private ChannelInterceptor[] interceptors; + S id(String id) { + this.id = id; + return _this(); + } + public S datatypes(Class... datatypes) { this.datatypes = datatypes; return _this(); @@ -44,12 +50,8 @@ public abstract class ChannelSpecSupport, C e return _this(); } - public S id(String id) { - this.id = id; - return _this(); - } - - public C get() { + @Override + protected C doGet() { this.channel.setDatatypes(this.datatypes); this.channel.setBeanName(this.id); if (this.interceptors != null) { @@ -58,9 +60,5 @@ public abstract class ChannelSpecSupport, C e return this.channel; } - @SuppressWarnings("unchecked") - protected S _this() { - return (S) this; - } } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java index 391ade7..a0bbba9 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java @@ -19,7 +19,6 @@ package org.springframework.integration.dsl.channel; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Executor; -import org.springframework.integration.dispatcher.LoadBalancingStrategy; import org.springframework.integration.store.MessageGroupStore; import org.springframework.messaging.Message; @@ -32,46 +31,82 @@ public final class MessageChannels { return new DirectChannelSpec(); } - public static DirectChannelSpec direct(LoadBalancingStrategy loadBalancingStrategy) { - return direct().loadBalancer(loadBalancingStrategy); + public static DirectChannelSpec direct(String id) { + return direct().id(id); } public static QueueChannelSpec queue() { return new QueueChannelSpec(); } + public static QueueChannelSpec queue(String id) { + return queue().id(id); + } + public static QueueChannelSpec queue(BlockingQueue> queue) { return new QueueChannelSpec(queue); } + public static QueueChannelSpec queue(String id, BlockingQueue> queue) { + return queue(queue).id(id); + } + public static QueueChannelSpec queue(Integer capacity) { return new QueueChannelSpec(capacity); } + public static QueueChannelSpec queue(String id, Integer capacity) { + return queue(capacity).id(id); + } + public static QueueChannelSpec.MessageStoreSpec queue(MessageGroupStore messageGroupStore, Object groupId) { return new QueueChannelSpec.MessageStoreSpec(messageGroupStore, groupId); } + public static QueueChannelSpec.MessageStoreSpec queue(String id, MessageGroupStore messageGroupStore, Object groupId) { + return queue(messageGroupStore, groupId).id(id); + } + public static ExecutorChannelSpec executor(Executor executor) { return new ExecutorChannelSpec(executor); } + public static ExecutorChannelSpec executor(String id, Executor executor) { + return executor(executor).id(id); + } + public static RendezvousChannelSpec rendezvous() { return new RendezvousChannelSpec(); } + public static RendezvousChannelSpec rendezvous(String id) { + return rendezvous().id(id); + } + public static PriorityChannelSpec priority() { return new PriorityChannelSpec(); } + public static PriorityChannelSpec priority(String id) { + return priority().id(id); + } + public static PublishSubscribeChannelSpec publishSubscribe() { return new PublishSubscribeChannelSpec(); } + public static PublishSubscribeChannelSpec publishSubscribe(String id) { + return publishSubscribe().id(id); + } + public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) { return new PublishSubscribeChannelSpec(executor); } + public static PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) { + return publishSubscribe(executor).id(id); + } + private MessageChannels() { } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java index 70f8c71..4ea9c17 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PriorityChannelSpec.java @@ -24,7 +24,7 @@ import org.springframework.messaging.Message; /** * @author Artem Bilan */ -public class PriorityChannelSpec extends ChannelSpecSupport { +public class PriorityChannelSpec extends MessageChannelSpec { private int capacity; @@ -41,9 +41,9 @@ public class PriorityChannelSpec extends ChannelSpecSupport { +public class PublishSubscribeChannelSpec extends MessageChannelSpec { PublishSubscribeChannelSpec() { this.channel = new PublishSubscribeChannel(); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java index c66197b..f3e9ff4 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java @@ -27,7 +27,7 @@ import org.springframework.messaging.Message; /** * @author Artem Bilan */ -public class QueueChannelSpec extends ChannelSpecSupport { +public class QueueChannelSpec extends MessageChannelSpec { protected BlockingQueue> queue; @@ -45,7 +45,7 @@ public class QueueChannelSpec extends ChannelSpecSupport { +public class RendezvousChannelSpec extends MessageChannelSpec { RendezvousChannelSpec() { this.channel = new RendezvousChannel(); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java new file mode 100644 index 0000000..bd6d5f3 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/Spec.java @@ -0,0 +1,41 @@ +/* + * Copyright 2014 the original author or authors. + * + * 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 org.springframework.integration.dsl.core; + +/** + * @author Artem Bilan + * @since 4.0 + */ +public abstract class Spec, T> { + + private volatile T target; + + public final T get() { + if (this.target == null) { + this.target = this.doGet(); + } + return this.target; + } + + protected abstract T doGet(); + + @SuppressWarnings("unchecked") + protected S _this() { + return (S) this; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java new file mode 100644 index 0000000..ef092ed --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java @@ -0,0 +1,94 @@ +/* + * Copyright 2014 the original author or authors. + * + * 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 org.springframework.integration.dsl.support; + +import java.util.Arrays; +import java.util.LinkedList; +import java.util.List; +import java.util.concurrent.Executor; + +import org.aopalliance.aop.Advice; + +import org.springframework.integration.dsl.core.Spec; +import org.springframework.integration.scheduling.PollerMetadata; +import org.springframework.integration.transaction.TransactionSynchronizationFactory; +import org.springframework.scheduling.Trigger; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.util.ErrorHandler; + +/** +* @author Artem Bilan +* @since 4.0 +*/ +public final class PollerSpec extends Spec { + + private final PollerMetadata pollerMetadata = new PollerMetadata(); + + private final List adviceChain = new LinkedList(); + + PollerSpec(Trigger trigger) { + this.pollerMetadata.setTrigger(trigger); + } + + public PollerSpec transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) { + pollerMetadata.setTransactionSynchronizationFactory(transactionSynchronizationFactory); + return this; + } + + public PollerSpec errorHandler(ErrorHandler errorHandler) { + pollerMetadata.setErrorHandler(errorHandler); + return this; + } + + public PollerSpec maxMessagesPerPoll(long maxMessagesPerPoll) { + pollerMetadata.setMaxMessagesPerPoll(maxMessagesPerPoll); + return this; + } + + public PollerSpec receiveTimeout(long receiveTimeout) { + pollerMetadata.setReceiveTimeout(receiveTimeout); + return this; + } + + public PollerSpec advice(Advice... advice) { + this.adviceChain.addAll(Arrays.asList(advice)); + return this; + } + + public PollerSpec transactional(PlatformTransactionManager transactionManager) { + return this.advice(new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource())); + } + + public PollerSpec taskExecutor(Executor taskExecutor) { + pollerMetadata.setTaskExecutor(taskExecutor); + return this; + } + + public PollerSpec sendTimeout(long sendTimeout) { + pollerMetadata.setSendTimeout(sendTimeout); + return this; + } + + @Override + protected PollerMetadata doGet() { + pollerMetadata.setAdviceChain(this.adviceChain); + return this.pollerMetadata; + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java index 14a31d6..8fdfe36 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Pollers.java @@ -16,24 +16,12 @@ package org.springframework.integration.dsl.support; -import java.util.Arrays; -import java.util.LinkedList; -import java.util.List; import java.util.TimeZone; -import java.util.concurrent.Executor; import java.util.concurrent.TimeUnit; -import org.aopalliance.aop.Advice; - -import org.springframework.integration.scheduling.PollerMetadata; -import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.support.CronTrigger; import org.springframework.scheduling.support.PeriodicTrigger; -import org.springframework.transaction.PlatformTransactionManager; -import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource; -import org.springframework.transaction.interceptor.TransactionInterceptor; -import org.springframework.util.ErrorHandler; /** * @author Artem Bilan @@ -77,62 +65,4 @@ public final class Pollers { private Pollers() { } - public static final class PollerSpec { - - private final PollerMetadata pollerMetadata = new PollerMetadata(); - - private final List adviceChain = new LinkedList(); - - private PollerSpec(Trigger trigger) { - this.pollerMetadata.setTrigger(trigger); - } - - public PollerSpec transactionSynchronizationFactory(TransactionSynchronizationFactory transactionSynchronizationFactory) { - pollerMetadata.setTransactionSynchronizationFactory(transactionSynchronizationFactory); - return this; - } - - public PollerSpec errorHandler(ErrorHandler errorHandler) { - pollerMetadata.setErrorHandler(errorHandler); - return this; - } - - public PollerSpec maxMessagesPerPoll(long maxMessagesPerPoll) { - pollerMetadata.setMaxMessagesPerPoll(maxMessagesPerPoll); - return this; - } - - public PollerSpec receiveTimeout(long receiveTimeout) { - pollerMetadata.setReceiveTimeout(receiveTimeout); - return this; - } - - public PollerSpec advice(Advice... advice) { - this.adviceChain.addAll(Arrays.asList(advice)); - return this; - } - - public PollerSpec transactional(PlatformTransactionManager transactionManager) { - return this.advice(new TransactionInterceptor(transactionManager, new MatchAlwaysTransactionAttributeSource())); - } - - public PollerSpec taskExecutor(Executor taskExecutor) { - pollerMetadata.setTaskExecutor(taskExecutor); - return this; - } - - public PollerSpec sendTimeout(long sendTimeout) { - pollerMetadata.setSendTimeout(sendTimeout); - return this; - } - - public PollerMetadata get() { - pollerMetadata.setAdviceChain(this.adviceChain); - return this.pollerMetadata; - } - - - - } - } diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index b9e4ff1..8d25af4 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -22,6 +22,7 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicInteger; import org.aopalliance.aop.Advice; @@ -42,7 +43,10 @@ import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.channel.QueueChannelSpec; +import org.springframework.integration.dsl.support.PollerSpec; import org.springframework.integration.dsl.support.Pollers; import org.springframework.integration.endpoint.MethodInvokingMessageSource; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; @@ -118,6 +122,16 @@ public class IntegrationFlowTests { @EnableIntegration public static class ContextConfiguration { + @Bean + public DirectChannelSpec inputChannel() { + return MessageChannels.direct(); + } + + @Bean + public QueueChannelSpec successChannel() { + return MessageChannels.queue(); + } + @Bean public MessageSource integerMessageSource() { MethodInvokingMessageSource source = new MethodInvokingMessageSource(); @@ -128,48 +142,51 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow flow1() { - return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100).get()) + return IntegrationFlows.from(this.integerMessageSource(), Pollers.fixedRate(100)) .transform("payload.toString()") - .channel(MessageChannels.queue().id("flow1QueueChannel").get()) + .channel(MessageChannels.queue("flow1QueueChannel")) .get(); } - @Bean - public DirectChannel inputChannel() { - return MessageChannels.direct().get(); - } - - @Bean - public QueueChannel successChannel() { - return MessageChannels.queue().get(); - } - @Bean(name = PollerMetadata.DEFAULT_POLLER_METADATA_BEAN_NAME) - public PollerMetadata poller() { - return Pollers.fixedRate(500).get(); + public PollerSpec poller() { + return Pollers.fixedRate(500); } + } + + @Configuration + public static class ContextConfiguration2 { + + @Autowired + @Qualifier("inputChannel") + private DirectChannel inputChannel; + + @Autowired + @Qualifier("successChannel") + private PollableChannel successChannel; + @Bean public Advice expressionAdvice() { ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(); advice.setOnSuccessExpression("payload"); - advice.setSuccessChannel(this.successChannel()); + advice.setSuccessChannel(this.successChannel); return advice; } @Bean public IntegrationFlow flow2() { - return IntegrationFlows.from(this.inputChannel()) + return IntegrationFlows.from(this.inputChannel) .filter(p -> p instanceof String, c -> c.id("filter")) .transform(Integer::parseInt) .transform(new PayloadSerializingTransformer(), c -> c.autoStartup(false).id("payloadSerializingTransformer")) - .channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue").get()) + .channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue")) .transform(new PayloadDeserializingTransformer()) + .channel(MessageChannels.executor("executor", Executors.newCachedThreadPool())) .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice())) .get(); } - } }