-
Notifications
You must be signed in to change notification settings - Fork 38.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
PartGenerator streaming mode mixing parts content, or part content not properly delivered when downstream part prefetch is 0 #27743
Comments
Resolving this complicated issue will take some time, with the possibility that we might need several iterations before getting it right. Scheduling for 6.0 M3 as a consequence. |
To be clear: depending on the resolution, we might back port the solution to this issue to the 5.3.x branch once resolved. |
I have a prototype implementation that works (in Groovy, but easily translatable in Java) : however, this impl is just managing the streaming case (not the non-streaming case). |
Sounds good. Please file a PR if you have some code that could be useful for us, referring back to this issue in the PR. |
We are considering to introduce a different way to handle streaming multipart support, because of inherent problems in the current model. See the description #28006. It would be great if you could share your thoughts in that issue. |
Hey, currently I need to build a upload endpoint that does not store data on this or holds the whole file in memory. This seems not to be possible right now due to this issue. When I activate streaming to true my controller never get's called.
Better said I thing the issue is on the emiPart which holds to what @djouvin said. |
Hello! Could you please provide your prototype? I'd like to have a look :) |
Here is the groovy code I used. |
Closing this issue, because we have deprecated |
Affects: \5.2.x
Affects: \5.3.x
When trying to use the
PartGenerator
in steaming mode (streaming = true), the documentation states that part content should be consumed in an ordered and serial fashion. We still end up with aFlux<Part>
, but of course we cannot use theflatMap
operator to trigger part content consumption since it would break this requirement.However, when using it, instead of raising an exception stating that parallel consumption of parts has been tried, the generator first silently mixes up the parts content (especially when very small parts are encountered, or when the part consumption is slower that the part reception/production), and a not explicit exception is finally thrown when trying to consume the ultimate or penultimate etc. (depending on how many parts have been silently skipped).
For example, if you have, say, a 20 KB part, then à 1 KB part and finally a 250 KB part, you will probably end up with the content of the 3rd part being sent as the 2nd part content, without notice, and when trying to consume the 3rd part, an exception will be thrown saying "Could not switch from STREAMING to STREAMING; current state: STREAMING".
If you try to use the
concatMap
operator, which is supposed to serialize mapper execution (and thus subscription to part content), the problems still appears because of the default prefetch of the operator (usually 32) : parts are still produced in advance, skipping small ones' content. So if you want to really serialize part content consumption, you have to useconcatMap
with no prefetch.However, when doing so, you are faced with another problem (which is a bug) : no part's body content (i.e. part content's databuffer) are produced at all, even after subscription and unbounded request.
My analysis of the problem
This is due to the fact that the current implementation (through the
requestToken()
method) does not account for the correct downstream request (given bysink.requestedFromDownstream()
) : only the downstream request from the parts' sink is accounted for, whereas when delivering part content, the content sink should be considered and not the part sink. Thus, two distinctrequestToken()
methods (or a parameterizedrequestToken(Sink sink)
) should be defined, to distinguish between part demand and content demand : for example, we could define therequestContent()
andrequestPart()
methods, with conditions pointing to different sinks. And therequestContent()
method should probably be defined at theStreamingState
inner class level, in order to point to the correct sink.Another problem is that a new intermediate state should be defined : currenty, there is only one state for part chaining and content delivery, the
StreamingState
(that we could rename toContentStreamingState
). But a transitional state, that we could callPartStreamingState
, should be defined to represent the fact that we are waiting for a new part to be requested from downstream, when all previous part content has been exhausted. The former would userequestContent()
and the latterrequestPart()
.Also, the part number, and/or a flag stating that the part is already passed or not, should be defined in the
StreamingState
inner class, to be able to produce a correct error message when the serial consumption of parts and part contents is not honored by the downstream subscriber. Whenever a part content is subscribed or requested, the flag or part number should be checked to ensure the consumer is subscribing or requesting the correct part (the current one), an dnot a revious one : if the consumer requests or subscribes too late to a part that has already been streamed, an explicit error message, specifying the part numbers, should be raised, like "Trying to subscribe a part (number #) that has already been completed (current part: #)". Also, theStreamingState
should display the part number in it'stoString()
method.Finally, the 'ContentStreamingState' should also use an inner queue to deliver the content body, only when content is requested. Currently, when a data buffer comprises multiple parts, the parts content is delivered unconditionnally : it should be enqueued instead and delivered only when requested by part content subscriber.
Note. We could also consider defining a new
StreamingPartGenerator
class to isolate the steaming behavior fromPartGenerator
, since this is a quite different behavior in nature than the "normal' one (saving parts to disc or memory before serving them).The text was updated successfully, but these errors were encountered: