berryware
commented
6 years ago Dalen,
It was thought about. You want to use the retrieve beans that takes a queuesize parameter and returns a CpoResultSet. I have include the javadoc below.
/**
- Retrieves the bean from the datasource. This method returns an Iterator immediately. The iterator will get filled
- asynchronously by the cpo framework. The framework will stop supplying the iterator with beans if the
- beanBufferSize is reached.
- If the consumer of the iterator is processing records faster than the framework is filling it, then the iterator
- will wait until it has data to provide.
- @param name The filter name which tells the datasource which beans should be returned. The name also signifies what
- data in the bean will be populated.
- @param criteria This is an bean that has been defined within the metadata of the datasource. If the class is not
- defined an exception will be thrown. If the bean does not exist in the datasource, an exception will be thrown.
- This bean is used to specify the arguments used to retrieve the collection of beans.
- @param result This is an bean that has been defined within the metadata of the datasource. If the class is not
- defined an exception will be thrown. If the bean does not exist in the datasource, an exception will be thrown.
- This bean is used to specify the bean type that will be returned in the collection.
- @param wheres A collection of CpoWhere beans that define the constraints that should be used when retrieving beans
- @param orderBy The CpoOrderBy bean that defines the order in which beans should be returned
- @param nativeExpressions Native expression that will be used to augment the expression stored in the meta data. This
- text will be embedded at run-time
- @param queueSize the maximum number of beans that the Iterator is allowed to cache. Once reached, the CPO
- framework will halt processing records from the datasource.
- @return An iterator that will be fed beans from the CPO framework.
-
@throws CpoException Thrown if there are errors accessing the datasource */ @Override public <T, C> CpoResultSet
retrieveBeans(String name, C criteria, T result, Collection wheres, Collection orderBy, Collection nativeExpressions, int queueSize) throws CpoException { CpoBlockingResultSet resultSet = new CpoBlockingResultSet<>(queueSize); RetrieverThread<T, C> retrieverThread = new RetrieverThread<>(name, criteria, result, wheres, orderBy, nativeExpressions, false, resultSet); retrieverThread.start(); return resultSet; }
The jUnits include a file called RetrieveBeanTest. There are several tests that called "testRetrieveBeansNoWaitSizeXX" that use the queuesize and then operate on the iterator. Here is an example:
public void testRetrieveBeansNoWaitSize20() {
String method = "testRetrieveBeansNoWaitSize20:";
CpoResultSet
try { ValueObject valObj = new ValueObjectBean(); crs = cpoAdapter.retrieveBeans(null, valObj, valObj, null, null, null, 20); logger.debug("Returned from retrieveBeans"); for (ValueObject vo : crs) { if (vo != null) { count++; } logger.debug("Retrieved Object #" + count); } assertTrue("Result size is " + count, count == al.size());
} catch (Exception e) { fail(method + e.getMessage()); } }
I hope this helps. I know with MySQL you need to provide specific connection string parameters for it to allow streaming/non-blocking. You may need to do this with oracle as well.
Dave
-----Original Message----- From: dkambur notifications@github.com To: synchronoss/cpo-api cpo-api@noreply.github.com Cc: Subscribed subscribed@noreply.github.com Sent: Mon, Nov 19, 2018 4:29 am Subject: [synchronoss/cpo-api] Streaming from CPO (#37)
Hi guys, Was there ever any thought put into streaming from CPO and how is this envisioned to work? On a particular project in SNCR, we have implemented streaming via JAXRS however one big issue is around getting actual data using CPO in blocks: namely, we implement multiple calls to cpo.retrieveBeans each to retrieve one block of data from Oracle whereas each block is filtered using rownums so that first request takes say rownum < 1000, second one is 1001 to 2000 etc. While this kind of works the way it was written, there are several issues:
if the underlying table changes (say by inserting or deleting a row), all rows read subsequently will be skewed so we'll end up with missing or duplicate rows it's not very fast as sending multiple queries, one after one, using rownum to filter upon specific block of data seems to be relatively slow it's very prone to errors in coding as all resulting columns need be specified in ORDER BY to make sure reads are consistent
If we were not using CPO, I'd read from a ResultSet first block, and maintain the same ResultSet in between calls for individual blocks but I'm not sure how this can be done in CPO world. Any thoughts or ideas you can share? Thanks, Dalen — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub, or mute the thread.
dkambur
Hi guys,
Was there ever any thought put into streaming from CPO and how is this envisioned to work?
On a particular project in SNCR, we have implemented streaming via JAXRS however one big issue is around getting actual data using CPO in blocks: namely, we implement multiple calls to cpo.retrieveBeans each to retrieve one block of data from Oracle whereas each block is filtered using rownums so that first request takes say rownum < 1000, second one is 1001 to 2000 etc.
While this kind of works the way it was written, there are several issues:
If we were not using CPO, I'd read from a ResultSet first block, and maintain the same ResultSet in between calls for individual blocks but I'm not sure how this can be done in CPO world.
Any thoughts or ideas you can share?
Thanks,
Dalen