Negotiate, install modifiers to protocols during handshake

InstalledProtocolsProcedure includes modifier protocols

Split modifier protocols and NettyPipelineBuilder into server/client.

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/EnterpriseCoreEditionModule.java

@@ -64,11 +64,12 @@
 import org.neo4j.causalclustering.messaging.Outbound;
 import org.neo4j.causalclustering.messaging.RaftOutbound;
 import org.neo4j.causalclustering.messaging.SenderService;
+import org.neo4j.causalclustering.protocol.ModifierProtocolInstaller;
 import org.neo4j.causalclustering.protocol.NettyPipelineBuilderFactory;
-import org.neo4j.causalclustering.protocol.Protocol;
 import org.neo4j.causalclustering.protocol.ProtocolInstaller;
-import org.neo4j.causalclustering.protocol.ProtocolInstallerRepository;
 import org.neo4j.causalclustering.protocol.handshake.HandshakeClientInitializer;
+import org.neo4j.causalclustering.protocol.Protocol;
+import org.neo4j.causalclustering.protocol.ProtocolInstallerRepository;
 import org.neo4j.causalclustering.protocol.handshake.ProtocolRepository;
 import org.neo4j.causalclustering.protocol.handshake.ProtocolStack;
 import org.neo4j.com.storecopy.StoreUtil;
@@ -120,6 +121,7 @@
 
 import static java.util.Collections.singletonList;
 import static org.neo4j.causalclustering.core.CausalClusteringSettings.raft_messages_log_path;
+import static org.neo4j.helpers.collection.Iterators.asSet;
 
 /**
  * This implementation of {@link org.neo4j.kernel.impl.factory.EditionModule} creates the implementations of services
@@ -234,11 +236,15 @@ public EnterpriseCoreEditionModule( final PlatformModule platformModule,
 
         long logThresholdMillis = config.get( CausalClusteringSettings.unknown_address_logging_throttle ).toMillis();
 
-        ProtocolRepository protocolRepository = new ProtocolRepository( Protocol.Protocols.values() );
+        ProtocolRepository<Protocol.ApplicationProtocol> applicationProtocolRepository = new ProtocolRepository<>( Protocol.ApplicationProtocols.values() );
+        ProtocolRepository<Protocol.ModifierProtocol> modifierProtocolRepository = new ProtocolRepository<>( Protocol.ModifierProtocols.values() );
         ProtocolInstallerRepository<ProtocolInstaller.Orientation.Client> protocolInstallerRepository =
-                new ProtocolInstallerRepository<>( singletonList( new RaftProtocolClientInstaller( logProvider, clientPipelineBuilderFactory ) ) );
-        HandshakeClientInitializer channelInitializer = new HandshakeClientInitializer( logProvider, protocolRepository, Protocol.Identifier.RAFT,
-                protocolInstallerRepository, config, clientPipelineBuilderFactory );
+                new ProtocolInstallerRepository<>(
+                        singletonList( new RaftProtocolClientInstaller.Factory( clientPipelineBuilderFactory, logProvider ) ),
+                        ModifierProtocolInstaller.allClientInstallers );
+        HandshakeClientInitializer channelInitializer = new HandshakeClientInitializer( applicationProtocolRepository,
+                Protocol.ApplicationProtocolIdentifier.RAFT, modifierProtocolRepository, asSet( Protocol.ModifierProtocolIdentifier.COMPRESSION ),
+                protocolInstallerRepository, clientPipelineBuilderFactory, config, logProvider );
         final SenderService raftSender = new SenderService( channelInitializer, logProvider );
         life.add( raftSender );
         this.clientInstalledProtocols = raftSender::installedProtocols;

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/RaftServerModule.java

@@ -39,6 +39,7 @@
 import org.neo4j.causalclustering.messaging.ComposableMessageHandler;
 import org.neo4j.causalclustering.messaging.LifecycleMessageHandler;
 import org.neo4j.causalclustering.messaging.LoggingInbound;
+import org.neo4j.causalclustering.protocol.ModifierProtocolInstaller;
 import org.neo4j.causalclustering.protocol.NettyPipelineBuilderFactory;
 import org.neo4j.causalclustering.protocol.Protocol;
 import org.neo4j.causalclustering.protocol.ProtocolInstaller;
@@ -84,15 +85,18 @@ class RaftServerModule
     private RaftServer createRaftServer( CoreServerModule coreServerModule,
             LifecycleMessageHandler<ReceivedInstantClusterIdAwareMessage<?>> messageHandlerChain )
     {
-        ProtocolRepository protocolRepository = new ProtocolRepository( Protocol.Protocols.values() );
+        ProtocolRepository<Protocol.ApplicationProtocol> applicationProtocolRepository = new ProtocolRepository<>( Protocol.ApplicationProtocols.values() );
+        ProtocolRepository<Protocol.ModifierProtocol> modifierProtocolRepository = new ProtocolRepository<>( Protocol.ModifierProtocols.values() );
 
         RaftMessageNettyHandler nettyHandler = new RaftMessageNettyHandler( logProvider );
-        RaftProtocolServerInstaller raftProtocolServerInstaller = new RaftProtocolServerInstaller( nettyHandler, pipelineBuilderFactory, logProvider );
+        RaftProtocolServerInstaller.Factory raftProtocolServerInstaller =
+                new RaftProtocolServerInstaller.Factory( nettyHandler, pipelineBuilderFactory, logProvider );
         ProtocolInstallerRepository<ProtocolInstaller.Orientation.Server> protocolInstallerRepository =
-                new ProtocolInstallerRepository<>( singletonList( raftProtocolServerInstaller ) );
+                new ProtocolInstallerRepository<>( singletonList( raftProtocolServerInstaller ), ModifierProtocolInstaller.allServerInstallers );
 
-        HandshakeServerInitializer handshakeServerInitializer = new HandshakeServerInitializer( logProvider, protocolRepository, Protocol.Identifier.RAFT,
-                protocolInstallerRepository, pipelineBuilderFactory );
+        HandshakeServerInitializer handshakeServerInitializer =
+                new HandshakeServerInitializer( logProvider, applicationProtocolRepository, modifierProtocolRepository,
+                        Protocol.ApplicationProtocolIdentifier.RAFT, protocolInstallerRepository, pipelineBuilderFactory );
         RaftServer raftServer = new RaftServer( handshakeServerInitializer, platformModule.config,
                 logProvider, platformModule.logging.getUserLogProvider() );
 

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/consensus/RaftProtocolClientInstaller.java

@@ -21,32 +21,63 @@
 
 import io.netty.channel.Channel;
 
+import java.util.List;
+import java.util.stream.Collectors;
+
 import org.neo4j.causalclustering.messaging.CoreReplicatedContentMarshal;
 import org.neo4j.causalclustering.messaging.marshalling.RaftMessageEncoder;
+import org.neo4j.causalclustering.protocol.ModifierProtocolInstaller;
 import org.neo4j.causalclustering.protocol.NettyPipelineBuilderFactory;
 import org.neo4j.causalclustering.protocol.Protocol;
 import org.neo4j.causalclustering.protocol.ProtocolInstaller;
+import org.neo4j.causalclustering.protocol.ProtocolInstaller.Orientation;
 import org.neo4j.logging.Log;
 import org.neo4j.logging.LogProvider;
 
-public class RaftProtocolClientInstaller extends ProtocolInstaller<ProtocolInstaller.Orientation.Client>
+public class RaftProtocolClientInstaller implements ProtocolInstaller<Orientation.Client>
 {
+    private static final Protocol.ApplicationProtocols APPLICATION_PROTOCOL = Protocol.ApplicationProtocols.RAFT_1;
+
+    public static class Factory extends ProtocolInstaller.Factory<Orientation.Client, RaftProtocolClientInstaller>
+    {
+        public Factory( NettyPipelineBuilderFactory clientPipelineBuilderFactory, LogProvider logProvider )
+        {
+            super( APPLICATION_PROTOCOL,
+                    modifiers -> new RaftProtocolClientInstaller( clientPipelineBuilderFactory, modifiers, logProvider ) );
+        }
+    }
+
+    private final List<ModifierProtocolInstaller<Orientation.Client>> modifiers;
     private final Log log;
     private final NettyPipelineBuilderFactory clientPipelineBuilderFactory;
 
-    public RaftProtocolClientInstaller( LogProvider logProvider, NettyPipelineBuilderFactory clientPipelineBuilderFactory )
+    public RaftProtocolClientInstaller( NettyPipelineBuilderFactory clientPipelineBuilderFactory, List<ModifierProtocolInstaller<Orientation.Client>> modifiers,
+            LogProvider logProvider )
     {
-        super( Protocol.Protocols.RAFT_1 );
+        this.modifiers = modifiers;
         this.log = logProvider.getLog( getClass() );
         this.clientPipelineBuilderFactory = clientPipelineBuilderFactory;
     }
 
     @Override
     public void install( Channel channel ) throws Exception
     {
-        clientPipelineBuilderFactory.create( channel, log )
+        clientPipelineBuilderFactory.client( channel, log )
+                .modify( modifiers )
                 .addFraming()
                 .add( "raft_encoder", new RaftMessageEncoder( new CoreReplicatedContentMarshal() ) )
                 .install();
     }
+
+    @Override
+    public Protocol.ApplicationProtocol applicationProtocol()
+    {
+        return APPLICATION_PROTOCOL;
+    }
+
+    @Override
+    public List<Protocol.ModifierProtocol> modifiers()
+    {
+        return modifiers.stream().map( ModifierProtocolInstaller::protocol ).collect( Collectors.toList() );
+    }
 }

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/consensus/RaftProtocolServerInstaller.java

@@ -23,36 +23,66 @@
 import io.netty.channel.ChannelInboundHandler;
 
 import java.time.Clock;
+import java.util.List;
+import java.util.stream.Collectors;
 
 import org.neo4j.causalclustering.messaging.CoreReplicatedContentMarshal;
 import org.neo4j.causalclustering.messaging.marshalling.RaftMessageDecoder;
+import org.neo4j.causalclustering.protocol.ModifierProtocolInstaller;
 import org.neo4j.causalclustering.protocol.NettyPipelineBuilderFactory;
 import org.neo4j.causalclustering.protocol.Protocol;
 import org.neo4j.causalclustering.protocol.ProtocolInstaller;
+import org.neo4j.causalclustering.protocol.ProtocolInstaller.Orientation;
 import org.neo4j.logging.Log;
 import org.neo4j.logging.LogProvider;
 
-public class RaftProtocolServerInstaller extends ProtocolInstaller<ProtocolInstaller.Orientation.Server>
+public class RaftProtocolServerInstaller implements ProtocolInstaller<Orientation.Server>
 {
+    private static final Protocol.ApplicationProtocols APPLICATION_PROTOCOL = Protocol.ApplicationProtocols.RAFT_1;
+
+    public static class Factory extends ProtocolInstaller.Factory<Orientation.Server, RaftProtocolServerInstaller>
+    {
+        public Factory( ChannelInboundHandler raftMessageHandler, NettyPipelineBuilderFactory pipelineBuilderFactory, LogProvider logProvider )
+        {
+            super( APPLICATION_PROTOCOL,
+                    modifiers -> new RaftProtocolServerInstaller( raftMessageHandler, pipelineBuilderFactory, modifiers, logProvider ) );
+        }
+    }
+
     private final ChannelInboundHandler raftMessageHandler;
     private final NettyPipelineBuilderFactory pipelineBuilderFactory;
+    private final List<ModifierProtocolInstaller<Orientation.Server>> modifiers;
     private final Log log;
 
-    public RaftProtocolServerInstaller( ChannelInboundHandler raftMessageHandler, NettyPipelineBuilderFactory pipelineBuilderFactory, LogProvider logProvider )
+    public RaftProtocolServerInstaller( ChannelInboundHandler raftMessageHandler, NettyPipelineBuilderFactory pipelineBuilderFactory,
+            List<ModifierProtocolInstaller<Orientation.Server>> modifiers, LogProvider logProvider )
     {
-        super( Protocol.Protocols.RAFT_1 );
         this.raftMessageHandler = raftMessageHandler;
         this.pipelineBuilderFactory = pipelineBuilderFactory;
+        this.modifiers = modifiers;
         this.log = logProvider.getLog( getClass() );
     }
 
     @Override
     public void install( Channel channel ) throws Exception
     {
-        pipelineBuilderFactory.create( channel, log )
+        pipelineBuilderFactory.server( channel, log )
+                .modify( modifiers )
                 .addFraming()
                 .add( "raft_decoder", new RaftMessageDecoder( new CoreReplicatedContentMarshal(), Clock.systemUTC() ) )
                 .add( "raft_handler", raftMessageHandler )
                 .install();
     }
+
+    @Override
+    public Protocol.ApplicationProtocol applicationProtocol()
+    {
+        return APPLICATION_PROTOCOL;
+    }
+
+    @Override
+    public List<Protocol.ModifierProtocol> modifiers()
+    {
+        return modifiers.stream().map( ModifierProtocolInstaller::protocol ).collect( Collectors.toList() );
+    }
 }

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/core/consensus/roles/Follower.java

@@ -149,7 +149,7 @@ public Outcome handle( RaftMessages.PreVote.Response response ) throws IOExcepti
         }
 
         @Override
-        public Outcome handle( RaftMessages.PruneRequest pruneRequest ) throws IOException
+        public Outcome handle( RaftMessages.PruneRequest pruneRequest )
         {
             Pruning.handlePruneRequest( outcome, pruneRequest );
             return outcome;
@@ -174,7 +174,7 @@ public Outcome handle( RaftMessages.Timeout.Election election ) throws IOExcepti
         }
 
         @Override
-        public Outcome handle( RaftMessages.Timeout.Heartbeat heartbeat ) throws IOException
+        public Outcome handle( RaftMessages.Timeout.Heartbeat heartbeat )
         {
             return outcome;
         }
@@ -242,7 +242,7 @@ public Outcome handle( RaftMessages.Timeout.Election election, Outcome outcome,
     private static class PreVoteUnsupportedRefusesToLead implements ElectionTimeoutHandler
     {
         @Override
-        public Outcome handle( RaftMessages.Timeout.Election election, Outcome outcome, ReadableRaftState ctx, Log log ) throws IOException
+        public Outcome handle( RaftMessages.Timeout.Election election, Outcome outcome, ReadableRaftState ctx, Log log )
         {
             log.info( "Election timeout triggered but refusing to be leader" );
             return outcome;
@@ -254,7 +254,7 @@ public Outcome handle( RaftMessages.Timeout.Election election, Outcome outcome,
     private static class PreVoteSupportedRefusesToLeadHandler implements ElectionTimeoutHandler
     {
         @Override
-        public Outcome handle( RaftMessages.Timeout.Election election, Outcome outcome, ReadableRaftState ctx, Log log ) throws IOException
+        public Outcome handle( RaftMessages.Timeout.Election election, Outcome outcome, ReadableRaftState ctx, Log log )
         {
             log.info( "Election timeout triggered but refusing to be leader" );
             Set<MemberId> memberIds = ctx.votingMembers();
@@ -295,7 +295,7 @@ public Outcome handle( RaftMessages.PreVote.Request request, Outcome outcome, Re
     private static class PreVoteRequestNoOpHandler implements PreVoteRequestHandler
     {
         @Override
-        public Outcome handle( RaftMessages.PreVote.Request request, Outcome outcome, ReadableRaftState ctx, Log log ) throws IOException
+        public Outcome handle( RaftMessages.PreVote.Request request, Outcome outcome, ReadableRaftState ctx, Log log )
         {
             return outcome;
         }
@@ -343,7 +343,7 @@ else if ( res.term() < ctx.term() || !res.voteGranted() )
     private static class PreVoteResponseNoOpHandler implements PreVoteResponseHandler
     {
         @Override
-        public Outcome handle( RaftMessages.PreVote.Response response, Outcome outcome, ReadableRaftState ctx, Log log ) throws IOException
+        public Outcome handle( RaftMessages.PreVote.Response response, Outcome outcome, ReadableRaftState ctx, Log log )
         {
             return outcome;
         }

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/discovery/procedures/InstalledProtocolsProcedure.java

@@ -21,8 +21,10 @@
 
 import java.util.Comparator;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import org.neo4j.causalclustering.protocol.Protocol;
 import org.neo4j.causalclustering.protocol.ProtocolInstaller;
 import org.neo4j.causalclustering.protocol.handshake.ProtocolStack;
 import org.neo4j.collection.RawIterator;
@@ -54,7 +56,8 @@ public InstalledProtocolsProcedure( Supplier<Stream<Pair<AdvertisedSocketAddress
                 .out( "orientation", Neo4jTypes.NTString )
                 .out( "remoteAddress", Neo4jTypes.NTString )
                 .out( "applicationProtocol", Neo4jTypes.NTString )
-                .out( "version", Neo4jTypes.NTInteger )
+                .out( "applicationProtocolVersion", Neo4jTypes.NTInteger )
+                .out( "modifierProtocols", Neo4jTypes.NTString )
                 .description( "Overview of installed protocols" )
                 .build() );
         this.clientInstalledProtocols = clientInstalledProtocols;
@@ -63,7 +66,7 @@ public InstalledProtocolsProcedure( Supplier<Stream<Pair<AdvertisedSocketAddress
 
     @Override
     public RawIterator<Object[],ProcedureException> apply(
-            Context ctx, Object[] input, ResourceTracker resourceTracker ) throws ProcedureException
+            Context ctx, Object[] input, ResourceTracker resourceTracker )
     {
         Stream<Object[]> outbound = toOutputRows( clientInstalledProtocols, ProtocolInstaller.Orientation.Client.OUTBOUND );
 
@@ -74,14 +77,34 @@ public RawIterator<Object[],ProcedureException> apply(
 
     private <T extends SocketAddress> Stream<Object[]> toOutputRows( Supplier<Stream<Pair<T,ProtocolStack>>> installedProtocols, String orientation )
     {
+        Comparator<Pair<T,ProtocolStack>> connectionInfoComparator = Comparator.comparing( ( Pair<T,ProtocolStack> entry ) -> entry.first().getHostname() )
+                .thenComparing( entry -> entry.first().getPort() );
+
         return installedProtocols.get()
-                .sorted( Comparator.comparing( entry -> entry.first().toString() ) )
-                .map( entry -> new Object[]
-                        {
-                                orientation,
-                                entry.first().toString(),
-                                entry.other().applicationProtocol().identifier(),
-                                (long) entry.other().applicationProtocol().version()
-                        } );
+                .sorted( connectionInfoComparator )
+                .map( entry -> buildRow( entry, orientation ) );
+    }
+
+    private <T extends SocketAddress> Object[] buildRow( Pair<T,ProtocolStack> connectionInfo, String orientation )
+    {
+        T socketAddress = connectionInfo.first();
+        ProtocolStack protocolStack = connectionInfo.other();
+        return new Object[]
+                {
+                    orientation,
+                    socketAddress.toString(),
+                    protocolStack.applicationProtocol().identifier(),
+                    (long) protocolStack.applicationProtocol().version(),
+                    modifierString( protocolStack )
+                };
+    }
+
+    private String modifierString( ProtocolStack protocolStack )
+    {
+        return protocolStack
+                .modifierProtocols()
+                .stream()
+                .map( Protocol.ModifierProtocol::friendlyName )
+                .collect( Collectors.joining( ",", "[", "]") );
     }
 }

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/messaging/MessageGate.java

@@ -71,7 +71,7 @@ public void userEventTriggered( ChannelHandlerContext ctx, Object evt ) throws E
     }
 
     @Override
-    public void write( ChannelHandlerContext ctx, Object msg, ChannelPromise promise ) throws Exception
+    public void write( ChannelHandlerContext ctx, Object msg, ChannelPromise promise )
     {
         if ( !gated.test( msg ) )
         {

enterprise/causal-clustering/src/main/java/org/neo4j/causalclustering/protocol/ClientNettyPipelineBuilder.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 2002-2018 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.causalclustering.protocol;
+
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.LengthFieldPrepender;
+
+import org.neo4j.logging.Log;
+
+public class ClientNettyPipelineBuilder extends NettyPipelineBuilder<ProtocolInstaller.Orientation.Client, ClientNettyPipelineBuilder>
+{
+    ClientNettyPipelineBuilder( ChannelPipeline pipeline, Log log )
+    {
+        super( pipeline, log );
+    }
+
+    @Override
+    public ClientNettyPipelineBuilder addFraming()
+    {
+        add( "frame_encoder", new LengthFieldPrepender( 4 ) );
+        return this;
+    }
+}