Hi I logged a ticket https://issues.apache.org/jira/browse/CAMEL-9843
And did a quick first implementation, a few things to irony out https://github.com/apache/camel/commit/18f3e27d847abe2fc39cc66cc4edd230cbecfc83 But it shows how you can setup it in a splitter with streaming mode. On Fri, Apr 8, 2016 at 1:41 AM, Ranx <[email protected]> wrote: > Claus, > Please forgive all the @ in here as that was the only way I could post this > to get around the spam filter. > > Just to get this working for me so I can start refactoring my code to accept > and process individual items from the unmarshaling of the multi-line beanio > beans I hacked this bit of code. I don't want to use it but it let's me > proceed until I can get something like this to work with the Camel file > endpoint. > > As you can see I'm injecting a seda queue in here to drop the individual > items on and a directory w@tcher to w@tch for incoming files. My unit test > copies a file over into the directory and I can see the results. I have to > put the exclusive lock mechanics in there or the bean reader would try to > read the file before it was truly available. But I don't want to replicate > all the mechanics that are already baked in and hardened in the Camel File > endpoint. > > Doing it this way though changes this from a pull mechanism to a > push/publish mechanism that I prefer. > > I have a couple of questions about how I might proceed with this. I'm fine > with reading the Beanio stream/definition from my blueprint bundle with > getResourceAsStream and foregoing the use of the Camel marsahaler itself. > If I do that, what mecahnics would I use to get a notification from the File > endpoint that a new file has been dropped into the directory. Just a > standard Camel Processor? > > You'd mentioned an iterator/reader object and noticed some in the BeanIO > code itself but I don't think I saw anything in Camel itself. > > I did try just chunking the raw text but the problem with that is if I read > in a chunk of 100 records the 100th might well fall in the middle of a > record. > > Thanks for any insights. > > @EndpointInject(uri = "seda:process") > private ProducerTemplate process; > > Logger logger = Logger.getLogger(AHSBatchReader.class); > > private StreamFactory factory; > private w@tchService w@tcher; > Path dir; > > public void init() throws IOException { > logger.info("Starting stream factory..."); > // create a StreamFactory > factory = StreamFactory.newInstance(); > // load the mapping file...switch to use getResourceAsStream > from bundle. > > factory.load("src/main/resources/dataformats/beanio-ahs-in.xml"); > > w@tcher = FileSystems.getDefault().neww@tchService(); > dir = FileSystems.getDefault().getPath("/test/in"); > dir.register(w@tcher, ENTRY_CREATE, ENTRY_DELETE, > ENTRY_MODIFY); > new Thread(this).start(); > } > > public void run() { > while (true) { > w@tchKey key; > try { > key = [email protected](); > > for (w@tchEvent<?> event : key.pollEvents()) { > [email protected] kind = > event.kind(); > w@tchEvent<Path> ev = cast(event); > Path name = ev.context(); > Path child = dir.resolve(name); > > // print out event > System.out.format("%s: %s\n", > event.kind().name(), child); > if (event.kind() == ENTRY_MODIFY) { > File file = child.toFile(); > boolean exclusivelyLocked = > false; > //Get an exclusive lock on the file if we can. If we can't then the system > is still using and it we delay start of beanio reading. > while (!exclusivelyLocked) { > try { > FileChannel > channel = new RandomAccessFile(file, "rw").getChannel(); > FileLock lock > = null; > try { > lock > = channel.tryLock(); > > > System.out.println(lock); > > System.out.println(lock.isShared()); > > System.out.println(lock.isValid()); > > exclusivelyLocked = lock.isValid() && !lock.isShared(); > } catch > (Exception e) { > // > File is open by someone else > > Thread.sleep(500); > } > //Release them now so that Beanio can grab them. > lock.close(); > > channel.close(); > } catch (Exception e) > { > } > } > //Start processing the beanio stream. > execute(file); > > } > } > > } catch (InterruptedException x) { > } > } > > } > > @SuppressWarnings("unchecked") > static <T> w@tchEvent<T> cast(w@tchEvent<?> event) { > return (w@tchEvent<T>) event; > } > > private void execute(File file) { > > BeanReader in = factory.createReader("myInputStream", > file.getAbsoluteFile()); > Object o; > while ((o = in.read()) != null) { > //Send these to the SEDA queue > process.sendBody(o); > } > > in.close(); > file.delete(); > > } > > } > > > > > -- > View this message in context: > http://camel.465427.n5.nabble.com/Beanio-stream-tp5780468p5780717.html > Sent from the Camel - Users mailing list archive at Nabble.com. -- Claus Ibsen ----------------- http://davsclaus.com @davsclaus Camel in Action 2: https://www.manning.com/ibsen2
