*/
package org.opendaylight.protocol.pcep.impl;
-import io.netty.util.concurrent.Future;
-
-import java.io.IOException;
+import com.google.common.base.Preconditions;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.util.concurrent.Promise;
import java.net.InetSocketAddress;
-import java.util.concurrent.ExecutionException;
-
-import org.opendaylight.protocol.framework.Dispatcher;
-import org.opendaylight.protocol.framework.ProtocolServer;
-import org.opendaylight.protocol.pcep.PCEPConnection;
-import org.opendaylight.protocol.pcep.PCEPConnectionFactory;
+import org.opendaylight.protocol.framework.AbstractDispatcher;
+import org.opendaylight.protocol.framework.SessionListenerFactory;
+import org.opendaylight.protocol.framework.SessionNegotiatorFactory;
import org.opendaylight.protocol.pcep.PCEPDispatcher;
-import org.opendaylight.protocol.pcep.PCEPSession;
-import org.opendaylight.protocol.pcep.PCEPSessionProposalFactory;
+import org.opendaylight.protocol.pcep.PCEPSessionListener;
+import org.opendaylight.protocol.pcep.spi.MessageRegistry;
+import org.opendaylight.tcpmd5.api.KeyMapping;
+import org.opendaylight.tcpmd5.netty.MD5ChannelFactory;
+import org.opendaylight.tcpmd5.netty.MD5ChannelOption;
+import org.opendaylight.tcpmd5.netty.MD5ServerChannelFactory;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Implementation of PCEPDispatcher.
*/
-public class PCEPDispatcherImpl implements PCEPDispatcher {
+public class PCEPDispatcherImpl extends AbstractDispatcher<PCEPSessionImpl, PCEPSessionListener> implements PCEPDispatcher {
+ private static final Logger LOG = LoggerFactory.getLogger(PCEPDispatcherImpl.class);
+ private final SessionNegotiatorFactory<Message, PCEPSessionImpl, PCEPSessionListener> snf;
+ private final MD5ServerChannelFactory<?> scf;
+ private final MD5ChannelFactory<?> cf;
+ private final PCEPHandlerFactory hf;
+ private KeyMapping keys;
- private final static Logger logger = LoggerFactory.getLogger(PCEPDispatcherImpl.class);
+ /**
+ * Creates an instance of PCEPDispatcherImpl, gets the default selector and opens it.
+ */
+ public PCEPDispatcherImpl(final MessageRegistry registry,
+ final SessionNegotiatorFactory<Message, PCEPSessionImpl, PCEPSessionListener> negotiatorFactory,
+ final EventLoopGroup bossGroup, final EventLoopGroup workerGroup) {
+ this(registry, negotiatorFactory, bossGroup, workerGroup, null, null);
+ }
- public static final int DEFAULT_MAX_UNKNOWN_MSG = 5;
+ /**
+ * Creates an instance of PCEPDispatcherImpl, gets the default selector and opens it.
+ */
+ public PCEPDispatcherImpl(final MessageRegistry registry,
+ final SessionNegotiatorFactory<Message, PCEPSessionImpl, PCEPSessionListener> negotiatorFactory,
+ final EventLoopGroup bossGroup, final EventLoopGroup workerGroup, final MD5ChannelFactory<?> cf,
+ final MD5ServerChannelFactory<?> scf) {
+ super(bossGroup, workerGroup);
+ this.cf = cf;
+ this.scf = scf;
+ this.snf = Preconditions.checkNotNull(negotiatorFactory);
+ this.hf = new PCEPHandlerFactory(registry);
+ }
- private int maxUnknownMessages = DEFAULT_MAX_UNKNOWN_MSG;
+ @Override
+ public synchronized ChannelFuture createServer(final InetSocketAddress address,
+ final SessionListenerFactory<PCEPSessionListener> listenerFactory) {
+ return createServer(address, null, listenerFactory);
+ }
- private final Dispatcher dispatcher;
+ @Override
+ public void close() {
+ }
- private final PCEPSessionProposalFactory proposalFactory;
+ @Override
+ protected void customizeBootstrap(final Bootstrap b) {
+ if (this.keys != null && !this.keys.isEmpty()) {
+ if (this.cf == null) {
+ throw new UnsupportedOperationException("No key access instance available, cannot use key mapping");
+ }
- /**
- * Creates an instance of PCEPDispatcherImpl, gets the default selector and opens it.
- *
- * @throws IOException if some error occurred during opening the selector
- */
- public PCEPDispatcherImpl(final Dispatcher dispatcher, final PCEPSessionProposalFactory proposalFactory) {
- this.dispatcher = dispatcher;
- this.proposalFactory = proposalFactory;
- }
+ LOG.debug("Adding MD5 keys {} to boostrap {}", this.keys, b);
+ b.channelFactory(this.cf);
+ b.option(MD5ChannelOption.TCP_MD5SIG, this.keys);
+ }
+ }
- @Override
- public Future<ProtocolServer> createServer(final InetSocketAddress address, final PCEPConnectionFactory connectionFactory) throws IOException {
- connectionFactory.setProposal(this.proposalFactory, address, 0);
- return this.dispatcher.createServer(address, connectionFactory, new PCEPSessionFactoryImpl(this.maxUnknownMessages));
- }
+ @Override
+ protected void customizeBootstrap(final ServerBootstrap b) {
+ if (this.keys != null && !this.keys.isEmpty()) {
+ if (this.scf == null) {
+ throw new UnsupportedOperationException("No key access instance available, cannot use key mapping");
+ }
- /**
- * Create client is used for mock purposes only.
- *
- * @throws ExecutionException
- * @throws InterruptedException
- */
- @Override
- public Future<? extends PCEPSession> createClient(final PCEPConnection connection) throws IOException {
- return this.dispatcher.createClient(connection, new PCEPSessionFactoryImpl(this.maxUnknownMessages));
- }
+ LOG.debug("Adding MD5 keys {} to boostrap {}", this.keys, b);
+ b.channelFactory(this.scf);
+ b.option(MD5ChannelOption.TCP_MD5SIG, this.keys);
+ }
- @Override
- public void setMaxUnknownMessages(final int limit) {
- this.maxUnknownMessages = limit;
- }
+ // Make sure we are doing round-robin processing
+ b.childOption(ChannelOption.MAX_MESSAGES_PER_READ, 1);
+ }
- public int getMaxUnknownMessages() {
- return this.maxUnknownMessages;
- }
+ @Override
+ public synchronized ChannelFuture createServer(final InetSocketAddress address, final KeyMapping keys,
+ final SessionListenerFactory<PCEPSessionListener> listenerFactory) {
+ this.keys = keys;
+ final ChannelFuture ret = super.createServer(address, new PipelineInitializer<PCEPSessionImpl>() {
+ @Override
+ public void initializeChannel(final SocketChannel ch, final Promise<PCEPSessionImpl> promise) {
+ ch.pipeline().addLast(PCEPDispatcherImpl.this.hf.getDecoders());
+ ch.pipeline().addLast("negotiator", PCEPDispatcherImpl.this.snf.getSessionNegotiator(listenerFactory, ch, promise));
+ ch.pipeline().addLast(PCEPDispatcherImpl.this.hf.getEncoders());
+ }
+ });
- public Dispatcher getDispatcher() {
- return this.dispatcher;
- }
+ this.keys = null;
+ return ret;
+ }
}
Code Review / bgpcep.git / blobdiff