apache camel - How to stop file transferring in spring-batch -


i have created spring-batch job reading files local directory , upload remote directory through ftp using camel-spring-batch. doing same using chunk. spring batch job configuration looks : 

<bean id="consumertemplate" class="org.apache.camel.impl.defaultconsumertemplate" init-method="start" destroy-method="stop">     <constructor-arg ref="camelcontext"/> </bean>  <bean id="producertemplate" class="org.apache.camel.impl.defaultproducertemplate" scope="step" init-method="start" destroy-method="stop">     <constructor-arg ref="camelcontext"/> </bean>  <bean id="localfilereader" class="com.camel.springbatch.reader.localfilereader" scope="step" destroy-method="stop">     <constructor-arg value="file:#{jobparameters['dirpath']}"/>     <constructor-arg ref="consumertemplate"/> </bean>  <bean id="ftpfilewriter" class="com.camel.springbatch.writer.ftpfilewriter" scope="step">     <constructor-arg ref="producertemplate"/>     <constructor-arg value="ftp://#{jobparameters['host']}?username=#{jobparameters['user']}&amp;password=#{jobparameters['password']}"/> </bean>

job configuration :

<batch:job id="ftpreadwrite">     <batch:step id="readfromlocalwritetoftp" next="readfromftpwritetolocal">         <batch:tasklet>             <batch:chunk reader="localfilereader" writer="ftpfilewriter"  commit-interval="5" />         </batch:tasklet>     </batch:step>

and "localfilereader" , "ftpfilewriter" looks :

import org.apache.camel.consumertemplate; import org.apache.camel.component.spring.batch.support.camelitemreader; import org.slf4j.logger; import org.slf4j.loggerfactory;  public class localfilereader extends camelitemreader { private logger log= loggerfactory.getlogger(this.getclass()); consumertemplate consumertemplate; string endpointuri;  public localfilereader(consumertemplate consumertemplate, string endpointuri) {     super(consumertemplate, endpointuri);     this.consumertemplate=consumertemplate;     this.endpointuri=endpointuri; }  @override public object read() throws exception {     object item = consumertemplate.receivebody(endpointuri);     return item; }

}

"ftp file writer"

import org.apache.camel.producertemplate; import org.apache.camel.component.spring.batch.support.camelitemwriter; import org.slf4j.logger; import org.slf4j.loggerfactory; import java.util.list; public class ftpfilewriter extends camelitemwriter { private logger log= loggerfactory.getlogger(this.getclass()); producertemplate producertemplate; string endpointuri; public ftpfilewriter(producertemplate producertemplate, string endpointuri) {     super(producertemplate, endpointuri);     this.producertemplate=producertemplate;     this.endpointuri=endpointuri; }  @override public void write(list items) throws exception {     system.out.println("************************writing item ftp "+items);     (object item : items) {         system.out.println("writing item [{}]..."+item);         producertemplate.sendbody(endpointuri, item);         log.debug("wrote item");     } } }

it works fine if have 5 file in local directory. read 5 file local directory , send writer , writer send ftp server commit-interval=5. if have 6 file in in local directory send first chunk of 5 file writer , again start reading remaining file , time there 1 file remaining. read 1 file , start waiting 4 files , never send writer. tried commit-interval=1 send 6 files server , again start waiting next file. here need stop process once file have been processed.

please me resolved issue...

from consumertemplate's javadoc receivebody waits until there response; need work timeout (check timeoutpolicy in spring-batch) or different way mark reader 'exhausted' (return null reader) stop reader reading


Comments

Post a Comment

Popular posts from this blog

basic authentication with http post params android -

i doing basic authentication passing username , password , using basicnamevaluepair sending post params response service. my method: public stringbuilder callservicehttppost(string username, string password, string type) { // create new httpclient , post header httpclient httpclient = new defaulthttpclient(); httppost httppost = new httppost(webservice + type); httpresponse response = null; stringbuilder total = new stringbuilder(); try { url url = new url(webservice + type); /*string base64encodedcredentials = base64.encodetostring((username + ":" + password).getbytes(), base64.url_safe | base64.no_wrap);*/ string base64encodedcredentials = "basic " + base64.encodetostring( (username + ":" + password).getbytes(), base64.no_wrap); httppost.setheader("auth...
Read more

vb.net - Virtual Keyboard commands -

i wondering difference between virtual these keyboard commands is: keyeventf_extendedkey , keyeventf_keyup is. everywhere have looked gives me description based off integers , not want know each of them does. you've tagged question vb.net, these have nothing @ vb.net. they're constants defined in windows header files, use win32 api functions. as far difference, can't tell looking @ values. individual values not particularly important, that's why named identifiers used. what's important used , documentation functions tells mean. the first one, keyeventf_extendedkey , used keybdinput structure (which used along e.g. sendinput function) pass information synthesized keyboard input. if flag used, means scan code should interpreted extended key. technically, means scan code preceded prefix byte value 224 (&he0 in hexadecimal notation). the second one, keyeventf_keyup , 1 of flags available use structure. means key being released (going up)...
Read more

c++ - End of file on pipe magic during open -

i have c++ application in starting process(wireshark) following. if (fp == null){ fp = popen(processpath, "r"); //processpath process want start if (!fp){ throw std::invalid_argument("cannot start process"); } fprintf(fp, d_msg);//d_msg input want provide process } else if(fp != null){ fprintf(fp, d_msg); } the problem when execute c++ application, start wireshark error end of file on pipe magic during open what should avoid that? also tried using mkfifo create named pipe , execute it. used this: if (fp == null){ system("mkfifo /tmp/mine.pcap"); fp = popen("wireshark -k -i /tmp/mine.pcap", "r"); if (!fp){ dout << "cannot start wireshark"<<std::endl; throw std::invalid_argument("cannot start wireshark"); } input = fopen("/tmp/mine.pcap", "wb"); fprintf(input , d_msg); fclose(input)...
Read more