From bdc5fa0dd0980eb9e077ae80644504705a24e035 Mon Sep 17 00:00:00 2001
From: Yannick Lecaillez <yannick.lecaillez@forgerock.com>
Date: Mon, 07 Nov 2016 13:59:40 +0000
Subject: [PATCH] OPENDJ-3179: Migrate LDAP Connection Handler to SDK Grizzly transport
---
opendj-grizzly/src/main/java/com/forgerock/opendj/grizzly/GrizzlyTransportProvider.java | 148 ++-----------------------------------------------
1 files changed, 6 insertions(+), 142 deletions(-)
diff --git a/opendj-grizzly/src/main/java/com/forgerock/opendj/grizzly/GrizzlyTransportProvider.java b/opendj-grizzly/src/main/java/com/forgerock/opendj/grizzly/GrizzlyTransportProvider.java
index 126f97b..6aa495a 100644
--- a/opendj-grizzly/src/main/java/com/forgerock/opendj/grizzly/GrizzlyTransportProvider.java
+++ b/opendj-grizzly/src/main/java/com/forgerock/opendj/grizzly/GrizzlyTransportProvider.java
@@ -15,42 +15,18 @@
*/
package com.forgerock.opendj.grizzly;
-import static com.forgerock.reactive.RxJavaStreams.streamFromPublisher;
-
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.Set;
import org.forgerock.opendj.grizzly.GrizzlyLDAPConnectionFactory;
import org.forgerock.opendj.grizzly.GrizzlyLDAPListener;
-import org.forgerock.opendj.io.ASN1Reader;
-import org.forgerock.opendj.io.AbstractLDAPMessageHandler;
-import org.forgerock.opendj.io.LDAP;
-import org.forgerock.opendj.io.LDAPReader;
-import org.forgerock.opendj.ldap.CommonLDAPOptions;
-import org.forgerock.opendj.ldap.DecodeException;
-import org.forgerock.opendj.ldap.DecodeOptions;
import org.forgerock.opendj.ldap.LDAPClientContext;
-import org.forgerock.opendj.ldap.LDAPClientContext.DisconnectListener;
import org.forgerock.opendj.ldap.LdapException;
-import org.forgerock.opendj.ldap.ResultCode;
-import org.forgerock.opendj.ldap.ServerConnection;
-import org.forgerock.opendj.ldap.ServerConnectionFactory;
-import org.forgerock.opendj.ldap.requests.AbandonRequest;
-import org.forgerock.opendj.ldap.requests.AddRequest;
-import org.forgerock.opendj.ldap.requests.CompareRequest;
-import org.forgerock.opendj.ldap.requests.DeleteRequest;
-import org.forgerock.opendj.ldap.requests.ExtendedRequest;
-import org.forgerock.opendj.ldap.requests.GenericBindRequest;
-import org.forgerock.opendj.ldap.requests.ModifyDNRequest;
-import org.forgerock.opendj.ldap.requests.ModifyRequest;
-import org.forgerock.opendj.ldap.requests.SearchRequest;
-import org.forgerock.opendj.ldap.requests.UnbindRequest;
-import org.forgerock.opendj.ldap.responses.ExtendedResult;
import org.forgerock.opendj.ldap.responses.Response;
import org.forgerock.opendj.ldap.spi.LDAPConnectionFactoryImpl;
import org.forgerock.opendj.ldap.spi.LDAPListenerImpl;
-import org.forgerock.opendj.ldap.spi.LdapMessages.LdapRawMessage;
+import org.forgerock.opendj.ldap.spi.LdapMessages.LdapRequestEnvelope;
import org.forgerock.opendj.ldap.spi.TransportProvider;
import org.forgerock.util.Function;
import org.forgerock.util.Options;
@@ -58,11 +34,6 @@
import com.forgerock.reactive.ReactiveHandler;
import com.forgerock.reactive.Stream;
-import io.reactivex.BackpressureStrategy;
-import io.reactivex.Flowable;
-import io.reactivex.FlowableEmitter;
-import io.reactivex.FlowableOnSubscribe;
-
/**
* Grizzly transport provider implementation.
*/
@@ -75,122 +46,15 @@
@Override
public LDAPListenerImpl getLDAPListener(final Set<InetSocketAddress> addresses,
- final ServerConnectionFactory<LDAPClientContext, Integer> factory, final Options options)
- throws IOException {
- return new GrizzlyLDAPListener(addresses, options,
- new Function<LDAPClientContext,
- ReactiveHandler<LDAPClientContext, LdapRawMessage, Stream<Response>>,
- LdapException>() {
- @Override
- public ReactiveHandler<LDAPClientContext, LdapRawMessage, Stream<Response>> apply(
- final LDAPClientContext clientContext) throws LdapException {
- return newHandler(clientContext, factory, options);
- }
- });
+ final Function<LDAPClientContext,
+ ReactiveHandler<LDAPClientContext, LdapRequestEnvelope, Stream<Response>>,
+ LdapException> factory,
+ final Options options) throws IOException {
+ return new GrizzlyLDAPListener(addresses, options, factory);
}
@Override
public String getName() {
return "Grizzly";
}
-
- private ReactiveHandler<LDAPClientContext, LdapRawMessage, Stream<Response>> newHandler(
- final LDAPClientContext clientContext, final ServerConnectionFactory<LDAPClientContext, Integer> factory,
- final Options options) throws LdapException {
- final ServerConnection<Integer> serverConnection = factory.handleAccept(clientContext);
- final ServerConnectionAdaptor<Integer> adapter = new ServerConnectionAdaptor<>(serverConnection);
- clientContext.onDisconnect(new DisconnectListener() {
- @Override
- public void exceptionOccurred(final LDAPClientContext context, final Throwable error) {
- serverConnection.handleConnectionError(error);
- }
-
- @Override
- public void connectionClosed(final LDAPClientContext context, final UnbindRequest unbindRequest) {
- serverConnection.handleConnectionClosed(0, unbindRequest);
- }
-
- @Override
- public void connectionDisconnected(final LDAPClientContext context, final ResultCode resultCode,
- final String diagnosticMessage) {
- serverConnection.handleConnectionDisconnected(resultCode, diagnosticMessage);
- }
- });
-
- final DecodeOptions decodeOptions = options.get(CommonLDAPOptions.LDAP_DECODE_OPTIONS);
- return new ReactiveHandler<LDAPClientContext, LdapRawMessage, Stream<Response>>() {
- @Override
- public Stream<Response> handle(final LDAPClientContext context,
- final LdapRawMessage rawRequest) throws Exception {
- final LDAPReader<ASN1Reader> reader = LDAP.getReader(rawRequest.getContent(), decodeOptions);
- return streamFromPublisher(Flowable.create(new FlowableOnSubscribe<Response>() {
- @Override
- public void subscribe(final FlowableEmitter<Response> emitter) throws Exception {
- reader.readMessage(new AbstractLDAPMessageHandler() {
- @Override
- public void abandonRequest(int messageID, AbandonRequest request)
- throws DecodeException, IOException {
- adapter.handleAbandon(messageID, request, emitter);
- }
-
- @Override
- public void addRequest(int messageID, AddRequest request)
- throws DecodeException, IOException {
- adapter.handleAdd(messageID, request, emitter);
- }
-
- @Override
- public void deleteRequest(final int messageID, final DeleteRequest request)
- throws DecodeException, IOException {
- adapter.handleDelete(messageID, request, emitter);
- }
-
- @Override
- public void bindRequest(int messageID, int version, GenericBindRequest request)
- throws DecodeException, IOException {
- adapter.handleBind(messageID, version, request, emitter);
- }
-
- @Override
- public void compareRequest(int messageID, CompareRequest request)
- throws DecodeException, IOException {
- adapter.handleCompare(messageID, request, emitter);
- }
-
- @Override
- public <R extends ExtendedResult> void extendedRequest(int messageID,
- ExtendedRequest<R> request) throws DecodeException, IOException {
- adapter.handleExtendedRequest(messageID, request, emitter);
- }
-
- @Override
- public void modifyDNRequest(int messageID, ModifyDNRequest request)
- throws DecodeException, IOException {
- adapter.handleModifyDN(messageID, request, emitter);
- }
-
- @Override
- public void modifyRequest(int messageID, ModifyRequest request)
- throws DecodeException, IOException {
- adapter.handleModify(messageID, request, emitter);
- }
-
- @Override
- public void searchRequest(int messageID, SearchRequest request)
- throws DecodeException, IOException {
- adapter.handleSearch(messageID, request, emitter);
- }
-
- @Override
- public void unbindRequest(int messageID, UnbindRequest request)
- throws DecodeException, IOException {
- serverConnection.handleConnectionClosed(messageID, request);
- }
- });
- emitter.onComplete();
- }
- }, BackpressureStrategy.ERROR));
- }
- };
- }
}
--
Gitblit v1.10.0