avoid duplicate sending and global notifier mode

pom.xml

@@ -66,7 +66,7 @@
     <dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpclient</artifactId>
-      <version>4.4</version>
+      <version>4.5.3</version>
     </dependency>
     <dependency>
       <groupId>redis.clients</groupId>
@@ -109,6 +109,16 @@
       <version>2.15</version>
       <optional>true</optional>
     </dependency>
+    <dependency>
+      <groupId>org.jenkins-ci.plugins</groupId>
+      <artifactId>conditional-buildstep</artifactId>
+      <version>1.3.6</version>
+    </dependency>
+    <dependency>
+      <groupId>org.jenkins-ci.plugins</groupId>
+      <artifactId>flexible-publish</artifactId>
+      <version>0.15.2</version>
+    </dependency>
     <dependency>
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
@@ -258,6 +268,40 @@
           </execution>
         </executions>
       </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-enforcer-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>display-info</id>
+            <goals>
+              <goal>display-info</goal>
+              <goal>enforce</goal>
+            </goals>
+            <phase>validate</phase>
+            <configuration>
+              <rules>
+                <requireUpperBoundDeps>
+                  <excludes>
+                    <exclude>org.codehaus.plexus:plexus-classworlds</exclude>
+                    <exclude>org.jenkins-ci.plugins:conditional-buildstep</exclude>
+                    <exclude>org.codehaus.plexus:plexus-utils</exclude>
+                    <exclude>com.google.guava:guava</exclude>
+                    <exclude>commons-logging:commons-logging</exclude>
+                    <exclude>com.google.code.findbugs:jsr305</exclude>
+                    <exclude>net.java.dev.jna:jna</exclude>
+                    <exclude>org.apache.ant:ant</exclude>
+                    <exclude>org.apache.maven:maven-embedder</exclude>
+                    <exclude>org.apache.maven:maven-core</exclude>
+                    <exclude>org.apache.maven:maven-aether-provider</exclude>
+                    <exclude>org.jenkins-ci.plugins:matrix-project</exclude>
+                  </excludes>
+                </requireUpperBoundDeps>
+              </rules>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
     </plugins>
   </build>
   <profiles>

src/main/java/jenkins/plugins/logstash/GloballyEnabledMode.java

@@ -0,0 +1,22 @@
+package jenkins.plugins.logstash;
+
+public enum GloballyEnabledMode
+{
+
+  // TODO: Introduce a DATAONLY mode, that allows to send a start event and an end event only without sending the log itself.
+  OFF("Do not send logs globally."),
+  LINEMODE("Send log line by line."),
+  NOTIFIERMODE("Send log at the end of the build.");
+
+  private String description;
+
+  private GloballyEnabledMode(String description)
+  {
+    this.description = description;
+  }
+
+  public String getDescrition()
+  {
+    return description;
+  }
+}

src/main/java/jenkins/plugins/logstash/LogstashConfiguration.java

@@ -39,9 +39,11 @@ public class LogstashConfiguration extends GlobalConfiguration
   private LogstashIndexer<?> logstashIndexer;
   private Boolean enabled;
   private boolean dataMigrated = false;
-  private boolean enableGlobally = false;
+  private transient boolean enableGlobally = false;
   private boolean milliSecondTimestamps = true;
   private transient LogstashIndexer<?> activeIndexer;
+  private GloballyEnabledMode globalMode;
+  private int maxLines = 1000;
 
   // a flag indicating if we're currently in the configure method.
   private transient boolean configuring = false;
@@ -63,6 +65,26 @@ public LogstashConfiguration()
     activeIndexer = logstashIndexer;
   }
 
+  public int getMaxLines()
+  {
+    return maxLines;
+  }
+
+  public void setMaxLines(int maxLines)
+  {
+    this.maxLines = maxLines;
+  }
+
+  public GloballyEnabledMode getGlobalMode()
+  {
+    return globalMode;
+  }
+
+  public void setGlobalMode(GloballyEnabledMode globalMode)
+  {
+    this.globalMode = globalMode;
+  }
+
   public boolean isEnabled()
   {
     return enabled == null ? false: enabled;
@@ -73,14 +95,23 @@ public void setEnabled(boolean enabled)
     this.enabled = enabled;
   }
 
+  @Deprecated
   public boolean isEnableGlobally()
   {
-    return enableGlobally;
+    return Objects.equals(globalMode, GloballyEnabledMode.LINEMODE);
   }
 
+  @Deprecated
   public void setEnableGlobally(boolean enableGlobally)
   {
-    this.enableGlobally = enableGlobally;
+    if (enableGlobally)
+    {
+      globalMode = GloballyEnabledMode.LINEMODE;
+    }
+    else
+    {
+      globalMode = GloballyEnabledMode.OFF;
+    }
   }
 
   public boolean isMilliSecondTimestamps()
@@ -230,6 +261,17 @@ public void migrateData()
       dataMigrated = true;
       save();
     }
+    if (globalMode == null)
+    {
+      if (enableGlobally == true)
+      {
+        globalMode = GloballyEnabledMode.LINEMODE;
+      }
+      else
+      {
+        globalMode = GloballyEnabledMode.OFF;
+      }
+    }
   }
 
   @Override
@@ -246,7 +288,7 @@ public boolean configure(StaplerRequest staplerRequest, JSONObject json) throws
       save();
       return true;
     }
-    
+
     configuring = true;
 
     // when we bind the stapler request we get a new instance of logstashIndexer.

src/main/java/jenkins/plugins/logstash/LogstashConsoleLogFilter.java

@@ -8,9 +8,9 @@
 
 import hudson.Extension;
 import hudson.console.ConsoleLogFilter;
-import hudson.model.AbstractBuild;
 import hudson.model.AbstractProject;
 import hudson.model.Run;
+import hudson.tasks.Publisher;
 
 @Extension(ordinal = 1000)
 public class LogstashConsoleLogFilter extends ConsoleLogFilter implements Serializable
@@ -19,69 +19,104 @@ public class LogstashConsoleLogFilter extends ConsoleLogFilter implements Serial
   private static final Logger LOGGER = Logger.getLogger(LogstashConsoleLogFilter.class.getName());
 
   private transient Run<?, ?> run;
+
   public LogstashConsoleLogFilter() {}
 
   public LogstashConsoleLogFilter(Run<?, ?> run)
   {
     this.run = run;
   }
+
   private static final long serialVersionUID = 1L;
 
+  /**
+   * {@inheritDoc}
+   *
+   */
   @Override
   public OutputStream decorateLogger(Run build, OutputStream logger) throws IOException, InterruptedException
   {
+
     LogstashConfiguration configuration = LogstashConfiguration.getInstance();
     if (!configuration.isEnabled())
     {
       LOGGER.log(Level.FINE, "Logstash is disabled. Logs will not be forwarded.");
       return logger;
     }
 
-    if (build != null && build instanceof AbstractBuild<?, ?>)
+
+    // A pipeline step uses the constructor which sets run.
+    if (run != null)
     {
-      if (isLogstashEnabled(build))
-      {
-        LogstashWriter logstash = getLogStashWriter(build, logger);
-        return new LogstashOutputStream(logger, logstash);
-      }
-      else
+      if (run.getAction(LogstashMarkerAction.class) != null)
       {
+        LOGGER.log(Level.FINEST, "Logstash is enabled globally. No need to decorate the logger another time for {0}",
+            run.toString());
         return logger;
       }
+      return getLogstashOutputStream(run, logger);
     }
-    if (run != null)
-    {
-      LogstashWriter logstash = getLogStashWriter(run, logger);
-      return new LogstashOutputStream(logger, logstash);
-    }
-    else
+
+    LogstashMarkerAction markerAction = new LogstashMarkerAction();
+    build.addAction(markerAction);
+
+    // Not pipeline step so @{code build} should be set.
+    if (isLogstashEnabled(build))
     {
-      return logger;
+      markerAction.setLineModeEnabled(true);
+      return getLogstashOutputStream(build, logger);
     }
+
+    return logger;
+  }
+
+  private LogstashOutputStream getLogstashOutputStream(Run<?, ?> run, OutputStream logger)
+  {
+    LogstashWriter logstash = getLogStashWriter(run, logger);
+    return new LogstashOutputStream(logger, logstash);
   }
 
   LogstashWriter getLogStashWriter(Run<?, ?> build, OutputStream errorStream)
   {
     return new LogstashWriter(build, errorStream, null, build.getCharset());
   }
 
-  private boolean isLogstashEnabled(Run<?, ?> build)
+  private boolean isLogstashEnabledGlobally()
   {
     LogstashConfiguration configuration = LogstashConfiguration.getInstance();
-    if (configuration.isEnableGlobally())
+    if (configuration.getGlobalMode() == GloballyEnabledMode.LINEMODE)
     {
+      LOGGER.log(Level.FINEST, "Line mode is enabled globally.");
       return true;
     }
+    return false;
+  }
+
+  private boolean isLogstashEnabled(Run<?, ?> build)
+  {
+    if (build == null)
+    {
+      return false;
+    }
 
     if (build.getParent() instanceof AbstractProject)
     {
       AbstractProject<?, ?> project = (AbstractProject<?, ?>)build.getParent();
-      if (project.getProperty(LogstashJobProperty.class) != null)
+      LogstashJobProperty property = project.getProperty(LogstashJobProperty.class);
+      if (property != null)
       {
-        return true;
+        LOGGER.log(Level.FINEST, "Property is set and disableGlobal is: " + property.isDisableGlobal());
+        return !property.isDisableGlobal();
+      }
+      else
+      {
+        if (PluginImpl.getLogstashNotifier(project) != null)
+        {
+          return false;
+        }
+        return isLogstashEnabledGlobally();
       }
     }
     return false;
   }
-
 }

src/main/java/jenkins/plugins/logstash/LogstashJobProperty.java

@@ -1,38 +1,40 @@
 package jenkins.plugins.logstash;
 
 import org.kohsuke.stapler.DataBoundConstructor;
-import org.kohsuke.stapler.StaplerRequest;
+import org.kohsuke.stapler.DataBoundSetter;
 
 import hudson.Extension;
 import hudson.model.AbstractProject;
 import hudson.model.Job;
-import hudson.model.JobProperty;
-import hudson.model.JobPropertyDescriptor;
-import net.sf.json.JSONObject;
+import jenkins.model.OptionalJobProperty;
 
 /**
  * This JobProperty is a marker to decide if logs should be sent to an indexer.
  *
  */
-public class LogstashJobProperty extends JobProperty<Job<?, ?>>
+public class LogstashJobProperty extends OptionalJobProperty<Job<?, ?>>
 {
 
+  private boolean disableGlobal;
+
   @DataBoundConstructor
   public LogstashJobProperty()
   {}
 
+  public boolean isDisableGlobal()
+  {
+    return disableGlobal;
+  }
+
+  @DataBoundSetter
+  public void setDisableGlobal(boolean disableGlobal)
+  {
+    this.disableGlobal = disableGlobal;
+  }
+
   @Extension
-  public static class DescriptorImpl extends JobPropertyDescriptor
+  public static class DescriptorImpl extends OptionalJobPropertyDescriptor
   {
-    @Override
-    public JobProperty<?> newInstance(StaplerRequest req, JSONObject formData) throws FormException
-    {
-      if (formData.containsKey("enable"))
-      {
-        return new LogstashJobProperty();
-      }
-      return null;
-    }
 
     @Override
     public String getDisplayName()