Simplified creation of DeserializerFactory

into common implementations in DeserializerFactories as to deduplicate these
implementations found throught main and test code and make that code more readable

community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/input/csv/CsvInput.java

@@ -34,6 +34,9 @@
 import org.neo4j.unsafe.impl.batchimport.input.InputRelationship;
 import org.neo4j.unsafe.impl.batchimport.input.csv.InputGroupsDeserializer.DeserializerFactory;
 
+import static org.neo4j.unsafe.impl.batchimport.input.csv.DeserializerFactories.defaultNodeDeserializer;
+import static org.neo4j.unsafe.impl.batchimport.input.csv.DeserializerFactories.defaultRelationshipDeserializer;
+
 /**
  * Provides {@link Input} from data contained in tabular/csv form. Expects factories for instantiating
  * the {@link CharSeeker} objects seeking values in the csv data and header factories for how to
@@ -108,13 +111,7 @@ public InputIterable<InputNode> nodes()
             @Override
             public InputIterator<InputNode> iterator()
             {
-                DeserializerFactory<InputNode> factory = (dataHeader, dataStream, decorator, validator) ->
-                {
-                        InputNodeDeserialization deserialization =
-                                new InputNodeDeserialization( dataHeader, dataStream, groups, idType.idsAreExternal() );
-                        return new InputEntityDeserializer<>( dataHeader, dataStream, config.delimiter(),
-                                deserialization, decorator, validator, badCollector );
-                };
+                DeserializerFactory<InputNode> factory = defaultNodeDeserializer( groups, config, idType, badCollector );
                 return new InputGroupsDeserializer<>( nodeDataFactory.iterator(), nodeHeaderFactory, config,
                         idType, maxProcessors, 1, factory, Validators.<InputNode>emptyValidator(), InputNode.class );
             }
@@ -135,13 +132,8 @@ public InputIterable<InputRelationship> relationships()
             @Override
             public InputIterator<InputRelationship> iterator()
             {
-                DeserializerFactory<InputRelationship> factory = (dataHeader, dataStream, decorator, validator) ->
-                {
-                        InputRelationshipDeserialization deserialization =
-                                new InputRelationshipDeserialization( dataHeader, dataStream, groups );
-                        return new InputEntityDeserializer<>( dataHeader, dataStream, config.delimiter(),
-                                deserialization, decorator, validator, badCollector );
-                };
+                DeserializerFactory<InputRelationship> factory =
+                        defaultRelationshipDeserializer( groups, config, idType, badCollector );
                 return new InputGroupsDeserializer<>( relationshipDataFactory.iterator(), relationshipHeaderFactory,
                         config, idType, maxProcessors, 1, factory, new InputRelationshipValidator(),
                         InputRelationship.class );

community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/input/csv/DeserializerFactories.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2002-2016 "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 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.unsafe.impl.batchimport.input.csv;
+
+import org.neo4j.unsafe.impl.batchimport.input.Collector;
+import org.neo4j.unsafe.impl.batchimport.input.Groups;
+import org.neo4j.unsafe.impl.batchimport.input.InputNode;
+import org.neo4j.unsafe.impl.batchimport.input.InputRelationship;
+import org.neo4j.unsafe.impl.batchimport.input.csv.InputGroupsDeserializer.DeserializerFactory;
+
+/**
+ * Common {@link DeserializerFactory} implementations.
+ */
+public class DeserializerFactories
+{
+    public static DeserializerFactory<InputNode> defaultNodeDeserializer(
+            Groups groups, Configuration config, IdType idType, Collector badCollector )
+    {
+        return (header,stream,decorator,validator) ->
+        {
+            InputNodeDeserialization deserialization =
+                    new InputNodeDeserialization( header, stream, groups, idType.idsAreExternal() );
+            return new InputEntityDeserializer<>( header, stream, config.delimiter(),
+                    deserialization, decorator, validator, badCollector );
+        };
+    }
+
+    public static DeserializerFactory<InputRelationship> defaultRelationshipDeserializer(
+            Groups groups, Configuration config, IdType idType, Collector badCollector )
+    {
+        return (header,stream,decorator,validator) ->
+        {
+                InputRelationshipDeserialization deserialization =
+                        new InputRelationshipDeserialization( header, stream, groups );
+                return new InputEntityDeserializer<>( header, stream, config.delimiter(),
+                        deserialization, decorator, validator, badCollector );
+        };
+    }
+}

community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/input/csv/InputGroupsDeserializer.java

@@ -45,7 +45,7 @@ class InputGroupsDeserializer<ENTITY extends InputEntity>
     private final IdType idType;
     private InputIterator<ENTITY> currentInput = new InputIterator.Empty<>();
     private long previousInputsCollectivePositions;
-    private int previousInputProcessors = 1;
+    private int previousInputProcessors;
     private boolean currentInputOpen;
     private final int maxProcessors;
     private final DeserializerFactory<ENTITY> factory;
@@ -60,15 +60,15 @@ InputEntityDeserializer<ENTITY> create( Header dataHeader, CharSeeker dataStream
     }
 
     InputGroupsDeserializer( Iterator<DataFactory<ENTITY>> dataFactory, Header.Factory headerFactory,
-            Configuration config, IdType idType, int maxProcessors, int processors, DeserializerFactory<ENTITY> factory,
-            Validator<ENTITY> validator, Class<ENTITY> entityClass )
+            Configuration config, IdType idType, int maxProcessors, int initialProcessors,
+            DeserializerFactory<ENTITY> factory, Validator<ENTITY> validator, Class<ENTITY> entityClass )
     {
         super( dataFactory );
         this.headerFactory = headerFactory;
         this.config = config;
         this.idType = idType;
         this.maxProcessors = maxProcessors;
-        this.previousInputProcessors = processors;
+        this.previousInputProcessors = initialProcessors;
         this.factory = factory;
         this.validator = validator;
         this.entityClass = entityClass;

community/kernel/src/main/java/org/neo4j/unsafe/impl/batchimport/input/csv/ParallelInputEntityDeserializer.java

@@ -68,7 +68,7 @@ public class ParallelInputEntityDeserializer<ENTITY extends InputEntity> extends
 
     @SuppressWarnings( "unchecked" )
     public ParallelInputEntityDeserializer( Data<ENTITY> data, Header.Factory headerFactory, Configuration config,
-            IdType idType, int maxProcessors, int processors, DeserializerFactory<ENTITY> factory,
+            IdType idType, int maxProcessors, int initialProcessors, DeserializerFactory<ENTITY> factory,
             Validator<ENTITY> validator, Class<ENTITY> entityClass )
     {
         // Reader of chunks, characters aligning to nearest newline
@@ -110,7 +110,7 @@ public ParallelInputEntityDeserializer( Data<ENTITY> data, Header.Factory header
                 return entities.toArray( (ENTITY[]) Array.newInstance( entityClass, entities.size() ) );
             },
             () -> dataHeader.clone() /*We need to clone the stateful header to each processing thread*/ );
-            processing.processors( processors - processing.processors( 0 ) );
+            processing.processors( initialProcessors - processing.processors( 0 ) );
 
             // Utility cursor which takes care of moving over processed results from chunk to chunk
             Supplier<ENTITY[]> batchSupplier = rebaseBatches( processing );

community/kernel/src/test/java/org/neo4j/unsafe/impl/batchimport/input/csv/InputGroupsDeserializerTest.java

@@ -32,6 +32,8 @@
 import org.neo4j.unsafe.impl.batchimport.input.Collector;
 import org.neo4j.unsafe.impl.batchimport.input.Groups;
 import org.neo4j.unsafe.impl.batchimport.input.InputNode;
+import org.neo4j.unsafe.impl.batchimport.input.csv.InputGroupsDeserializer.DeserializerFactory;
+
 import static java.util.Arrays.asList;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -43,6 +45,7 @@
 import static org.neo4j.unsafe.impl.batchimport.input.InputEntityDecorators.NO_NODE_DECORATOR;
 import static org.neo4j.unsafe.impl.batchimport.input.csv.Configuration.COMMAS;
 import static org.neo4j.unsafe.impl.batchimport.input.csv.DataFactories.defaultFormatNodeFileHeader;
+import static org.neo4j.unsafe.impl.batchimport.input.csv.DeserializerFactories.defaultNodeDeserializer;
 import static org.neo4j.unsafe.impl.batchimport.input.csv.IdType.INTEGER;
 
 public class InputGroupsDeserializerTest
@@ -103,15 +106,10 @@ public void shouldCoordinateGroupCreationForParallelProcessing() throws Exceptio
         IdType idType = IdType.INTEGER;
         Collector badCollector = mock( Collector.class );
         Configuration config = lowBufferSize( COMMAS, false );
+        DeserializerFactory<InputNode> factory = defaultNodeDeserializer( groups, config, idType, badCollector );
         try ( InputGroupsDeserializer<InputNode> deserializer = new InputGroupsDeserializer<>(
                 data.iterator(), defaultFormatNodeFileHeader(), config, idType,
-                processors, processors, (header,stream,decorator,validator) ->
-                {
-                    InputNodeDeserialization deserialization =
-                            new InputNodeDeserialization( header, stream, groups, idType.idsAreExternal() );
-                    return new InputEntityDeserializer<>( header, stream, config.delimiter(),
-                            deserialization, decorator, validator, badCollector );
-                }, Validators.<InputNode>emptyValidator(), InputNode.class ) )
+                processors, processors, factory, Validators.<InputNode>emptyValidator(), InputNode.class ) )
         {
             // WHEN
             count( deserializer );