1. Introduced automatic watch reactivation on log changes and server restarts

2. Prevented records doubling on sequential watching toggling
3. Tuned visual look of composite records
4. Lifted composite logs to the top of choice list
5. Changed christmas substrate to classic one

Author: Toparvion

build.gradle

@@ -41,7 +41,7 @@ repositories {
 
 dependencies {
     // Backend compile deps
-    compile("org.springframework.boot:spring-boot-devtools:${springBootVersion}")
+//    compile("org.springframework.boot:spring-boot-devtools:${springBootVersion}")
     compile("org.springframework.boot:spring-boot-starter-web:${springBootVersion}")
     compile("org.springframework.boot:spring-boot-starter-actuator:${springBootVersion}")
     compile("org.springframework.boot:spring-boot-starter-websocket:${springBootVersion}")

config/application-win.yaml

@@ -18,8 +18,9 @@ choices:
   - group: MT
     pathBase: C:\lang\analog\log-samples\
     plainLogs:
-      - core.log as "$f - $g"
+      - core.log as "static $f - $g"
       - pppinfo.log as "$f - $g"
+      - generated\core.log as "generated $f - $g"
     compositeLogs:
       - path: generated\bankplus.log
         node: angara
@@ -44,10 +45,7 @@ choices:
     compositeLogs:
       - path: bankplus.log
         title: $f - $g
-        selected: false
-        encoding: CP-1251
         timestamp: dd.MM.yy HH:mm:ss,SSS
       - path: logSample.log
         title: $f - $g
-        selected: false
         timestamp: dd.MM.yy HH:mm:ss,SSS

src/main/java/tech/toparvion/analog/model/TrackingRequest.java

@@ -16,12 +16,14 @@ public class TrackingRequest implements Serializable {
   private final String timestampFormat;
   private final String nodeName;
   private final String uid;
+  private final boolean isTailNeeded;
 
-  public TrackingRequest(String logFullPath, @Nullable String timestampFormat, String nodeName, String uid) {
+  public TrackingRequest(String logFullPath, @Nullable String timestampFormat, String nodeName, String uid, boolean isTailNeeded) {
     this.logFullPath = logFullPath;
     this.timestampFormat = timestampFormat;
     this.nodeName = nodeName;
     this.uid = uid;
+    this.isTailNeeded = isTailNeeded;
   }
 
   public String getLogFullPath() {
@@ -42,6 +44,10 @@ public String getUid() {
     return uid;
   }
 
+  public boolean isTailNeeded() {
+    return isTailNeeded;
+  }
+
   /**
    * A request with no timestamp format is considered 'plain' as it cannot be involved into complex aggregating
    * tracking logic and thus is suitable for plain old tracking only.
@@ -58,6 +64,7 @@ public String toString() {
         ", timestampFormat='" + timestampFormat + '\'' +
         ", nodeName='" + nodeName + '\'' +
         ", uid='" + uid + '\'' +
+        ", isTailNeeded=" + isTailNeeded +
         '}';
   }
 

src/main/java/tech/toparvion/analog/remote/agent/TailingFlowProvider.java

@@ -53,9 +53,10 @@ public TailingFlowProvider(TimestampExtractor timestampExtractor,
    * The core method for building AnaLog dynamic behavior. Creates and returns an integration flow for watching given
    * log. No duplicate flow checking is done inside.
    * @param logPath full path to log file to tail
+   * @param isTailNeeded should 'tail' include last several lines of the log?
    * @return a new tailing flow
    */
-  IntegrationFlow provideAggregatingFlow(String logPath) {
+  IntegrationFlow provideAggregatingFlow(String logPath, boolean isTailNeeded) {
     // each tailing flow must have its own instance of correlationProvider as it is stateful and not thread-safe
     CorrelationIdHeaderEnricher correlationProvider = new CorrelationIdHeaderEnricher();
     // each tailing flow must have its own instance of sequenceProvider as it is stateful and not thread-safe
@@ -79,7 +80,7 @@ sequence number (much like the one built in FileSplitter) and provided by dedica
     return IntegrationFlows
         .from(tailAdapter(new File(logPath))
             .id("tailSource:"+logPath)
-            .nativeOptions(tailSpecificsProvider.getCompositeTailNativeOptions())
+            .nativeOptions(tailSpecificsProvider.getCompositeTailNativeOptions(isTailNeeded))
             .fileDelay(tailSpecificsProvider.getAttemptsDelay())
             .enableStatusReader(true))   // to receive events of log rotation, etc.
         .enrichHeaders(e -> e.headerFunction(LOG_TIMESTAMP_VALUE__HEADER, timestampExtractor::extractTimestamp))
@@ -92,11 +93,11 @@ sequence number (much like the one built in FileSplitter) and provided by dedica
         .get();
   }
 
-  IntegrationFlow providePlainFlow(String logPath) {
+  IntegrationFlow providePlainFlow(String logPath, boolean isTailNeeded) {
     return IntegrationFlows
         .from(tailAdapter(new File(logPath))
             .id("tailSource:"+logPath)
-            .nativeOptions(tailSpecificsProvider.getPlainTailNativeOptions())
+            .nativeOptions(tailSpecificsProvider.getPlainTailNativeOptions(isTailNeeded))
             .fileDelay(tailSpecificsProvider.getAttemptsDelay())
             .enableStatusReader(true))   // to receive events log rotation, etc.
         .aggregate(aggregatorSpec -> aggregatorSpec

src/main/java/tech/toparvion/analog/remote/agent/TrackingService.java

@@ -74,8 +74,8 @@ public TrackingService(@SuppressWarnings("SpringJavaInjectionPointsAutowiringIns
   /**
    * Initiates tracking process for the log specified in {@code request}: <ol>
    *   <li>Checks if this is a duplicate registration (the request for the same log came from the same watcher);</li>
-   *   <li>Finds or {@link TailingFlowProvider#provideAggregatingFlow(java.lang.String) creates} tailing flow (alongside
-   *   with the {@link TimestampExtractor#registerNewTimestampFormat(java.lang.String, java.lang.String) registration}
+   *   <li>Finds or {@link TailingFlowProvider#provideAggregatingFlow(String, boolean) creates} tailing flow (alongside
+   *   with the {@link TimestampExtractor#registerNewTimestampFormat(String, String) registration}
    *   of the specified timestamp format);</li>
    *   <li>Creates new {@code RmiOutboundGateway} capable of sending messages to the {@code watcherAddress} and makes
    *   it a subscriber for the tailing flow output channel;</li>
@@ -110,7 +110,7 @@ void registerWatcher(TrackingRequest request, InetSocketAddress watcherAddress)
 
     } else if (!request.isPlain()) {
       IntegrationFlowRegistration trackingRegistration = flowContext
-          .registration(trackingFlowProvider.provideAggregatingFlow(logPath))
+          .registration(trackingFlowProvider.provideAggregatingFlow(logPath, request.isTailNeeded()))
           .autoStartup(true)
           .register();
       trackingRegistry.put(logPath, trackingRegistration.getId());
@@ -121,7 +121,7 @@ void registerWatcher(TrackingRequest request, InetSocketAddress watcherAddress)
 
     } else {
       IntegrationFlowRegistration trackingRegistration = flowContext
-          .registration(trackingFlowProvider.providePlainFlow(logPath))
+          .registration(trackingFlowProvider.providePlainFlow(logPath, request.isTailNeeded()))
           .autoStartup(true)
           .register();
       trackingRegistry.put(logPath, trackingRegistration.getId());

src/main/java/tech/toparvion/analog/remote/server/MetaDataSender.java

@@ -35,7 +35,7 @@ public MetaDataSender(SimpMessagingTemplate messagingTemplate, TailSpecificsProv
     this.tailSpecificsProvider = tailSpecificsProvider;
   }
 
-  public void sendMetaData(Message<?> metaMessage) {
+  void sendMetaData(Message<?> metaMessage) {
     // extract header values in order to include them into metadata being sent
     String uid = metaMessage.getHeaders().get(LOG_CONFIG_ENTRY_UID__HEADER, String.class);
     String sourceNode = metaMessage.getHeaders().get(SOURCE_NODE__HEADER, String.class);

src/main/java/tech/toparvion/analog/remote/websocket/WebSocketEventListener.java

@@ -72,10 +72,8 @@ public void onConnected(SessionConnectedEvent event) {
   public void onSubscribe(SessionSubscribeEvent event) {
     // First let's extract all the necessary info about new watching from the subscribe request
     StompHeaderAccessor headers = StompHeaderAccessor.wrap(event.getMessage());
-    List<String> rawHeaderValue = headers.getNativeHeader("isPlain");
-    assert (rawHeaderValue != null) && (rawHeaderValue.size() == 1)
-        : "'isPlain' header of SUBSCRIBE command is absent or malformed";
-    Boolean isPlain = Boolean.valueOf(rawHeaderValue.get(0));
+    boolean isPlain = getBooleanNativeHeader(headers, "isPlain");
+    boolean isTailNeeded = getBooleanNativeHeader(headers, "isTailNeeded");
     String sessionId = headers.getSessionId();
     String destination = headers.getDestination();
     assert (destination != null) && destination.startsWith(WEBSOCKET_TOPIC_PREFIX)
@@ -98,7 +96,7 @@ public void onSubscribe(SessionSubscribeEvent event) {
         // 1. Ensure that all RMI registration channels are created
         ensureRegistrationChannelsCreated(logConfig);
         // 2. Register the tracking on specified nodes
-        switchTracking(logConfig, true);
+        startTrackingOnServer(logConfig, isTailNeeded);
         // 3. Remember the tracking in the registry
         registry.addEntry(logConfig, sessionId);
         log.info("New tracking for log '{}' has started within session id={}.", logConfig.getUid(), sessionId);
@@ -108,6 +106,9 @@ public void onSubscribe(SessionSubscribeEvent event) {
         ServerFailure failure = new ServerFailure(e.getMessage(), now());
         messagingTemplate.convertAndSend(WEBSOCKET_TOPIC_PREFIX + logConfig.getUid(),
             failure, singletonMap(MESSAGE_TYPE_HEADER, MessageType.FAILURE));
+        // TODO Научиться по аналогии с этим отправлять сообщение об успешной настройке подписки, для чего превратить
+        // serverFailure в более общий тип сообщения. При получении этого типа убирать на клиенте прелоадер,
+        // выставленный перед отправкой запроса на подписку.
       }
 
     } else {    // i.e. if there are watching sessions already in registry
@@ -141,11 +142,11 @@ private void stopTrackingIfNeeded(Message<byte[]> message, boolean isUnsubscribi
     // check if there is no such session(s)
     if (fellows == null) {
       if (isUnsubscribing) {      // in case of unsubscribing it is incorrect situation
-        throw new IllegalStateException("No registered session(s) found for sessionId=" + sessionId);
+        log.warn("No registered session(s) found for sessionId={}", sessionId);
       } else {                    // but in case of disconnecting it is quite right
         log.info("No registered session found for sessionId={}.", sessionId);
-        return;
       }
+      return;
     }
     // check if there any other session(s) left (may require synchronization on registry object)
     if (fellows.size() > 1) {
@@ -157,13 +158,30 @@ private void stopTrackingIfNeeded(Message<byte[]> message, boolean isUnsubscribi
     // in case it was the latest session watching that log we should stop the tracking
     LogConfigEntry watchingLog = registry.findLogConfigEntryBy(sessionId);
     log.debug("No sessions left watching log '{}'. Will deactivate the tracking...", watchingLog.getUid());
-    switchTracking(watchingLog, false);
+    stopTrackingOnServer(watchingLog);
     // now that the log is not tracked anymore we need to remove it from the registry
     registry.removeEntry(watchingLog);
     log.info("Current node has unregistered itself from tracking log '{}' as there is no watching sessions anymore.",
         watchingLog.getUid());
   }
 
+  private boolean getBooleanNativeHeader(StompHeaderAccessor headers, String name) {
+    List<String> rawHeaderValue = headers.getNativeHeader(name);
+    assert (rawHeaderValue != null) && (rawHeaderValue.size() == 1)
+        : format("'%s' header of SUBSCRIBE command is absent or malformed", name);
+    return Boolean.valueOf(rawHeaderValue.get(0));
+  }
+
+  /**
+   * When watching request comes for a plain log AnaLog does not tries to find corresponding log config entry.
+   * Instead it just creates new ('artificial') entry and then works with it only. This approach allows AnaLog to
+   * watch arbitrary plain logs independently of its configuration. Particularly, it means that a user can set any
+   * path into AnaLog's address line and start to watch it the same way as if it was pre-configured as a choice
+   * variant in AnaLog configuration.
+   *
+   * @param path full path of log file to watch for
+   * @return newly created log config entry for the specified path
+   */
   @NotNull
   private LogConfigEntry createPlainLogConfigEntry(String path) {
     LogConfigEntry artificialEntry = new LogConfigEntry();
@@ -206,13 +224,22 @@ private void ensureRegistrationChannelsCreated(LogConfigEntry matchingEntry) {
         });
   }
 
-  private void switchTracking(LogConfigEntry logConfigEntry, boolean isOn) {
+  private void stopTrackingOnServer(LogConfigEntry logConfigEntry) {
+    switchTrackingOnServer(logConfigEntry, false, false);
+  }
+
+  private void startTrackingOnServer(LogConfigEntry logConfigEntry, boolean isTailNeeded) {
+    switchTrackingOnServer(logConfigEntry, true, isTailNeeded);
+  }
+
+  private void switchTrackingOnServer(LogConfigEntry logConfigEntry, boolean isOn, boolean isTailNeeded) {
+    assert !(isTailNeeded && !isOn) : "isTailNeeded flag should not be raised when switching the tracking off";
     ClusterNode myselfNode = clusterProperties.getMyselfNode();
     String fullPath = buildFullPath(logConfigEntry);
     String nodeName = nvls(logConfigEntry.getNode(), myselfNode.getName());
 
     // send registration request for the main entry
-    TrackingRequest primaryRequest = new TrackingRequest(fullPath, logConfigEntry.getTimestamp(), nodeName, logConfigEntry.getUid());
+    TrackingRequest primaryRequest = new TrackingRequest(fullPath, logConfigEntry.getTimestamp(), nodeName, logConfigEntry.getUid(), isTailNeeded);
     log.debug("Switching {} the registration by PRIMARY request: {}", isOn ? "ON":"OFF", primaryRequest);
     remoteGateway.switchRegistration(primaryRequest, isOn);
 
@@ -224,7 +251,8 @@ private void switchTracking(LogConfigEntry logConfigEntry, boolean isOn) {
             normalizePath(included.getPath()),
             included.getTimestamp(),
             nvls(included.getNode(), myselfNode.getName()),
-            logConfigEntry.getUid());
+            logConfigEntry.getUid(),
+            isTailNeeded);
         log.debug("Switching {} the registration by INCLUDED request: {}", isOn ? "ON":"OFF", includedRequest);
         remoteGateway.switchRegistration(includedRequest, isOn);
 

src/main/java/tech/toparvion/analog/service/LogChoicesProvider.java

@@ -50,8 +50,8 @@ List<LogChoice> provideLogChoices() {
 
   private Stream<LogChoice> flattenGroup(ChoiceGroup group) {
     Set<LogChoice> choices = new LinkedHashSet<>();
-    choices.addAll(processPlainLogs(group));
     choices.addAll(processCompositeLogs(group));
+    choices.addAll(processPlainLogs(group));
     choices.addAll(processScanDir(group));
     return choices.stream();
   }

src/main/java/tech/toparvion/analog/service/tail/GnuCoreUtilsTailSpecificsProvider.java

@@ -21,13 +21,15 @@ public static boolean matches(String idString) {
   }
 
   @Override
-  public String getCompositeTailNativeOptions() {
-    return "-F -n 20";     // follow file name with retries, starting with the very end of the file
+  public String getCompositeTailNativeOptions(boolean includePreviousLines) {
+    // follow file name with retries, starting with either the very EOF or some offset before the EOF
+    return "-F -n " + (includePreviousLines ? "20" : "0");
   }
 
   @Override
-  public String getPlainTailNativeOptions() {
-    return "-F -n 45";     // follow file name with retries, starting from some offset before the EOF
+  public String getPlainTailNativeOptions(boolean includePreviousLines) {
+    // follow file name with retries, starting with either the very EOF or some offset before the EOF
+    return "-F -n " + (includePreviousLines ? "45" : "0");
   }
 
   @Override

src/main/java/tech/toparvion/analog/service/tail/SolarisTailSpecificsProvider.java

@@ -2,6 +2,8 @@
 
 import tech.toparvion.analog.model.TailEventType;
 
+import static java.lang.String.format;
+
 /**
  * tail's specifics for Solaris OS family. Actually there are different tail implementations for various Solaris
  * distributions (SunOS, IllumOs, etc) but AnaLog doesn't need to distinguish them so far.
@@ -18,13 +20,13 @@ public static boolean matches(String idfString) {
   }
 
   @Override
-  public String getCompositeTailNativeOptions() {
-    return "-20f";     // -F option is not supported on SunOS
+  public String getCompositeTailNativeOptions(boolean includePreviousLines) {
+    return format("-%sf", includePreviousLines?"20":"0");     // -F option is not supported on SunOS
   }
 
   @Override
-  public String getPlainTailNativeOptions() {
-    return "-45f";     // -F option is not supported on SunOS
+  public String getPlainTailNativeOptions(boolean includePreviousLines) {
+    return format("-%sf", includePreviousLines?"45":"0");     // -F option is not supported on SunOS
   }
 
   @Override