Validate length of flux content matches with Content-Length header

[anuchandy]

Is your feature request related to a problem? Please describe.
Storage SDK has blob upload methods where user can provide flux emitting the content and length. The length become Content-Length header. A validation is missing that ensure length of data emitted by flux is equal to length param.

Describe the solution you'd like
instead of passing the flux to reactor-netty directly, append operators that does this check and pass updated flux to reactor-netty.

Describe alternatives you've considered
Adding this validation in storage SDK side is possible but with lot of code duplication since we have multiple upload methods.

Information Checklist
Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report

• [ X] Description Added
• [X ] Expected solution specified

[anuchandy]

When this validation fails throw UnexpectedLengthException

/**
 * Indicates that a stream emitted an unexpected number of bytes.
 */
public class UnexpectedLengthException extends IllegalStateException {
    private final long bytesRead;
    private final long bytesExpected;

    /**
     * Creates an UnexpectedLengthException.
     * @param message the message
     * @param bytesRead the number of bytes actually read in the stream
     * @param bytesExpected the number of bytes expected to be read in the stream
     */
    public UnexpectedLengthException(String message, long bytesRead, long bytesExpected) {
        super(message);
        this.bytesRead = bytesRead;
        this.bytesExpected = bytesExpected;
    }

    /**
     * @return the number of bytes actually read in the stream
     */
    public long bytesRead() {
        return bytesRead;
    }

    /**
     * @return the number of bytes expected to be read in the stream
     */
    public long bytesExpected() {
        return bytesExpected;
    }
}

[jianghaolu]

@joshfree this will unblock many Storage tests.

[sima-zhu]

Verified the current structure.
RestProxy is the front door in azure core which can take in the request body.
Will add the validation in RestProxy.