Skip to content

Commit cdbfc59

Browse files
mp911demp911de
committed
Add support to consume return values from stored procedures.
[closes #199] Signed-off-by: Mark Paluch <mpaluch@vmware.com>
1 parent bda0d23 commit cdbfc59

15 files changed

+611
-270
lines changed

‎README.md

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,15 +15,16 @@ This driver provides the following features:
1515
* Simple execution of SQL batches (direct and cursored execution)
1616
* Execution of parametrized statements (direct and cursored execution)
1717
* Extensive type support (including `TEXT`, `VARCHAR(MAX)`, `IMAGE`, `VARBINARY(MAX)` and national variants, see below for exceptions)
18+
* Execution of stored procedures
1819

1920
Next steps:
2021

21-
* Execution of stored procedures
2222
* Add support for TVP and UDTs
2323

2424
## Code of Conduct
2525

26-
This project is governed by the [R2DBC Code of Conduct](https://github.com/r2dbc/.github/blob/main/CODE_OF_CONDUCT.adoc). By participating, you are expected to uphold this code of conduct. Please report unacceptable behavior to [info@r2dbc.io](mailto:info@r2dbc.io).
26+
This project is governed by the [R2DBC Code of Conduct](https://github.com/r2dbc/.github/blob/main/CODE_OF_CONDUCT.adoc). By participating, you are expected to uphold this code of conduct. Please
27+
report unacceptable behavior to [info@r2dbc.io](mailto:info@r2dbc.io).
2728

2829
## Getting Started
2930

‎src/main/java/io/r2dbc/mssql/DefaultMssqlResult.java

Lines changed: 31 additions & 160 deletions
Original file line numberDiff line numberDiff line change
@@ -20,14 +20,14 @@
2020
import io.r2dbc.mssql.client.ConnectionContext;
2121
import io.r2dbc.mssql.codec.Codecs;
2222
import io.r2dbc.mssql.message.token.AbstractDoneToken;
23-
import io.r2dbc.mssql.message.token.AbstractInfoToken;
2423
import io.r2dbc.mssql.message.token.ColumnMetadataToken;
2524
import io.r2dbc.mssql.message.token.ErrorToken;
2625
import io.r2dbc.mssql.message.token.NbcRowToken;
2726
import io.r2dbc.mssql.message.token.ReturnValue;
2827
import io.r2dbc.mssql.message.token.RowToken;
2928
import io.r2dbc.mssql.util.Assert;
3029
import io.r2dbc.spi.R2dbcException;
30+
import io.r2dbc.spi.Readable;
3131
import io.r2dbc.spi.Result;
3232
import io.r2dbc.spi.Row;
3333
import io.r2dbc.spi.RowMetadata;
@@ -44,8 +44,6 @@
4444
import java.util.function.Predicate;
4545

4646
/**
47-
* TODO: Revisit segment-based result consumption as we need to materialize {@link Segment} for filter(…) and ReturnValue processing.
48-
* <p>
4947
* {@link Result} of query results.
5048
*
5149
* @author Mark Paluch
@@ -97,8 +95,7 @@ static MssqlResult toResult(String sql, ConnectionContext context, Codecs codecs
9795

9896
LOGGER.debug(context.getMessage("Creating new result"));
9997

100-
// return new MssqlResult(sql, context, codecs, messages, expectReturnValues);
101-
return MssqlSegmentResult.toResult(sql, context, codecs, messages, expectReturnValues);
98+
return new DefaultMssqlResult(sql, context, codecs, messages, expectReturnValues);
10299
}
103100

104101
@Override
@@ -144,14 +141,26 @@ public Mono<Integer> getRowsUpdated() {
144141
}
145142

146143
@Override
147-
public <T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> f) {
144+
public <T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> mappingFunction) {
145+
Assert.requireNonNull(mappingFunction, "Mapping function must not be null");
146+
return doMap(true, false, readable -> {
147+
Row row = (Row) readable;
148+
return mappingFunction.apply(row, row.getMetadata());
149+
});
150+
}
148151

149-
Assert.requireNonNull(f, "Mapping function must not be null");
152+
@Override
153+
public <T> Publisher<T> map(Function<? super Readable, ? extends T> mappingFunction) {
154+
Assert.requireNonNull(mappingFunction, "Mapping function must not be null");
155+
return doMap(true, true, mappingFunction);
156+
}
157+
158+
private <T> Flux<T> doMap(boolean rows, boolean outparameters, Function<? super Readable, ? extends T> mappingFunction) {
150159

151160
Flux<T> mappedReturnValues = Flux.empty();
152161
Flux<io.r2dbc.mssql.message.Message> messages = this.messages;
153162

154-
if (this.expectReturnValues) {
163+
if (this.expectReturnValues && outparameters) {
155164

156165
List<ReturnValue> returnValues = new ArrayList<>();
157166

@@ -160,23 +169,22 @@ public <T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> f) {
160169
if (message instanceof ReturnValue) {
161170
returnValues.add((ReturnValue) message);
162171
}
163-
});
172+
}).filter(it -> !(it instanceof ReturnValue));
164173

165174
mappedReturnValues = Flux.defer(() -> {
166175

167176
if (returnValues.size() != 0) {
168-
return Flux.just(MssqlReturnValues.toReturnValues(this.codecs, returnValues));
169-
}
170-
171-
return Flux.empty();
172177

173-
}).handle((row, sink) -> {
178+
MssqlReturnValues mssqlReturnValues = MssqlReturnValues.toReturnValues(this.codecs, returnValues);
174179

175-
try {
176-
sink.next(f.apply(row, row.getMetadata()));
177-
} finally {
178-
row.release();
180+
try {
181+
return Flux.just(mappingFunction.apply(mssqlReturnValues));
182+
} finally {
183+
mssqlReturnValues.release();
184+
}
179185
}
186+
187+
return Flux.empty();
180188
});
181189
}
182190

@@ -200,7 +208,7 @@ public <T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> f) {
200208
return;
201209
}
202210

203-
if (message.getClass() == RowToken.class || message.getClass() == NbcRowToken.class) {
211+
if (rows && (message.getClass() == RowToken.class || message.getClass() == NbcRowToken.class)) {
204212

205213
MssqlRowMetadata rowMetadata = this.rowMetadata;
206214

@@ -211,7 +219,7 @@ public <T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> f) {
211219

212220
MssqlRow row = MssqlRow.toRow(this.codecs, (RowToken) message, rowMetadata);
213221
try {
214-
sink.next(f.apply(row, row.getMetadata()));
222+
sink.next(mappingFunction.apply(row));
215223
} finally {
216224
row.release();
217225
}
@@ -253,150 +261,13 @@ public <T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> f) {
253261
}
254262

255263
@Override
256-
public Result filter(Predicate<Segment> filter) {
257-
258-
Flux<io.r2dbc.mssql.message.Message> filteredMessages = this.messages.filter(message -> {
259-
260-
if (message.getClass() == ColumnMetadataToken.class) {
261-
262-
ColumnMetadataToken token = (ColumnMetadataToken) message;
263-
264-
if (token.hasColumns()) {
265-
this.rowMetadata = MssqlRowMetadata.create(this.codecs, token);
266-
}
267-
return true;
268-
}
269-
270-
if (message.getClass() == RowToken.class || message.getClass() == NbcRowToken.class) {
271-
272-
MssqlRowMetadata rowMetadata = this.rowMetadata;
273-
274-
if (rowMetadata == null) {
275-
return false;
276-
}
277-
278-
MssqlRow row = MssqlRow.toRow(this.codecs, (RowToken) message, rowMetadata);
279-
280-
boolean result = filter.test(row);
281-
282-
if (!result) {
283-
row.release();
284-
}
285-
286-
return result;
287-
}
288-
289-
if (message instanceof AbstractInfoToken) {
290-
return filter.test(createMessage((AbstractInfoToken) message));
291-
}
292-
293-
if (message instanceof AbstractDoneToken) {
294-
295-
AbstractDoneToken doneToken = (AbstractDoneToken) message;
296-
if (doneToken.hasCount()) {
297-
298-
return filter.test(doneToken);
299-
}
300-
}
301-
return true;
302-
});
303-
304-
return new DefaultMssqlResult(this.sql, this.context, this.codecs, filteredMessages, this.expectReturnValues);
264+
public MssqlResult filter(Predicate<Segment> filter) {
265+
return MssqlSegmentResult.toResult(this.sql, this.context, this.codecs, this.messages, this.expectReturnValues).filter(filter);
305266
}
306267

307268
@Override
308269
public <T> Flux<T> flatMap(Function<Segment, ? extends Publisher<? extends T>> mappingFunction) {
309-
310-
return this.messages
311-
.flatMap(message -> {
312-
313-
if (message instanceof AbstractDoneToken) {
314-
315-
AbstractDoneToken doneToken = (AbstractDoneToken) message;
316-
if (doneToken.hasCount()) {
317-
318-
if (DEBUG_ENABLED) {
319-
LOGGER.debug(this.context.getMessage("Incoming row count: {}"), doneToken);
320-
}
321-
322-
return mappingFunction.apply(doneToken);
323-
}
324-
}
325-
326-
if (message.getClass() == ColumnMetadataToken.class) {
327-
328-
ColumnMetadataToken token = (ColumnMetadataToken) message;
329-
330-
if (!token.hasColumns()) {
331-
return Mono.empty();
332-
}
333-
334-
if (DEBUG_ENABLED) {
335-
LOGGER.debug(this.context.getMessage("Result column definition: {}"), message);
336-
}
337-
338-
this.rowMetadata = MssqlRowMetadata.create(this.codecs, token);
339-
}
340-
341-
if (message.getClass() == RowToken.class || message.getClass() == NbcRowToken.class) {
342-
343-
MssqlRowMetadata rowMetadata = this.rowMetadata;
344-
345-
if (rowMetadata == null) {
346-
return Mono.error(new IllegalStateException("No MssqlRowMetadata available"));
347-
}
348-
349-
MssqlRow row = MssqlRow.toRow(this.codecs, (RowToken) message, rowMetadata);
350-
351-
try {
352-
return Flux.from(mappingFunction.apply(row)).doFinally(it -> row.release());
353-
} catch (RuntimeException e) {
354-
row.release();
355-
throw e;
356-
}
357-
}
358-
359-
if (message instanceof AbstractInfoToken) {
360-
return mappingFunction.apply(createMessage((AbstractInfoToken) message));
361-
}
362-
363-
ReferenceCountUtil.release(message);
364-
365-
return Mono.empty();
366-
});
367-
}
368-
369-
private Message createMessage(AbstractInfoToken message) {
370-
371-
ErrorDetails errorDetails = ExceptionFactory.createErrorDetails(message);
372-
373-
return new Message() {
374-
375-
@Override
376-
public R2dbcException exception() {
377-
return ExceptionFactory.createException(message, DefaultMssqlResult.this.sql);
378-
}
379-
380-
@Override
381-
public int errorCode() {
382-
return (int) errorDetails.getNumber();
383-
}
384-
385-
@Override
386-
public String sqlState() {
387-
return errorDetails.getStateCode();
388-
}
389-
390-
@Override
391-
public String message() {
392-
return errorDetails.getMessage();
393-
}
394-
395-
@Override
396-
public Severity severity() {
397-
return message instanceof ErrorToken ? Severity.ERROR : Severity.INFO;
398-
}
399-
};
270+
return MssqlSegmentResult.toResult(this.sql, this.context, this.codecs, this.messages, this.expectReturnValues).flatMap(mappingFunction);
400271
}
401272

402273
}

‎src/main/java/io/r2dbc/mssql/MssqlColumnMetadata.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import io.r2dbc.mssql.util.Assert;
2323
import io.r2dbc.spi.ColumnMetadata;
2424
import io.r2dbc.spi.Nullability;
25+
import io.r2dbc.spi.OutParameterMetadata;
2526
import io.r2dbc.spi.Type;
2627

2728
import javax.annotation.Nonnull;
@@ -31,7 +32,7 @@
3132
*
3233
* @author Mark Paluch
3334
*/
34-
public final class MssqlColumnMetadata implements ColumnMetadata {
35+
public final class MssqlColumnMetadata implements ColumnMetadata, OutParameterMetadata {
3536

3637
private final Decodable decodable;
3738

‎src/main/java/io/r2dbc/mssql/MssqlResult.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,13 +44,13 @@ public interface MssqlResult extends Result {
4444
* {@inheritDoc}
4545
*/
4646
@Override
47-
<T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> f);
47+
<T> Flux<T> map(BiFunction<Row, RowMetadata, ? extends T> mappingFunction);
4848

4949
/**
5050
* {@inheritDoc}
5151
*/
5252
@Override
53-
Result filter(Predicate<Segment> filter);
53+
MssqlResult filter(Predicate<Segment> filter);
5454

5555
/**
5656
* {@inheritDoc}

‎src/main/java/io/r2dbc/mssql/MssqlReturnValues.java

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -19,13 +19,13 @@
1919
import io.netty.buffer.ByteBuf;
2020
import io.netty.util.ReferenceCounted;
2121
import io.r2dbc.mssql.codec.Codecs;
22+
import io.r2dbc.mssql.message.Message;
2223
import io.r2dbc.mssql.message.token.ReturnValue;
2324
import io.r2dbc.mssql.message.token.RowToken;
2425
import io.r2dbc.mssql.message.type.SqlServerType;
2526
import io.r2dbc.mssql.util.Assert;
26-
import io.r2dbc.spi.Result;
27+
import io.r2dbc.spi.OutParameters;
2728
import io.r2dbc.spi.Row;
28-
import io.r2dbc.spi.RowMetadata;
2929
import reactor.util.annotation.Nullable;
3030

3131
import java.util.List;
@@ -38,7 +38,7 @@
3838
* @see #release()
3939
* @see ReferenceCounted
4040
*/
41-
final class MssqlReturnValues implements Row, Result.Data {
41+
final class MssqlReturnValues implements OutParameters, Message {
4242

4343
private static final int STATE_ACTIVE = 0;
4444

@@ -52,7 +52,7 @@ final class MssqlReturnValues implements Row, Result.Data {
5252

5353
private volatile int state = STATE_ACTIVE;
5454

55-
MssqlReturnValues(Codecs codecs, List<ReturnValue> returnValues, MssqlReturnValuesMetadata metadata) {
55+
private MssqlReturnValues(Codecs codecs, List<ReturnValue> returnValues, MssqlReturnValuesMetadata metadata) {
5656

5757
this.codecs = codecs;
5858
this.metadata = metadata;
@@ -83,11 +83,6 @@ public MssqlReturnValuesMetadata getMetadata() {
8383
return this.metadata;
8484
}
8585

86-
@Override
87-
public RowMetadata metadata() {
88-
return this.metadata;
89-
}
90-
9186
@Override
9287
public <T> T get(int index, Class<T> type) {
9388

0 commit comments

Comments
 (0)