Refactor OriginsInventory.

[mikkokar]

• Use QueueDrainingEventProcessor to ensure lockless consistency in face of concurrent updates.
• Improve origin state management for added/modified/removed monitored origins.

[kvosper]

Result of registry reload?

[kvosper]

Convention would use this style: * @param <S> state type rather than full sentences for @param, @return @throws etc.

[kvosper]

The number of interfaces here may indicate that this class is taking on too many responsibilities, but we can look into that later.

[kvosper]

Why not just use Collections.emptyMap as that does not require an external dependency.

[kvosper]

add or set?

[mikkokar]

Renamed to setOriginsEventHandler.

[kvosper]

Suggested refactoring to encapsulate concept of origin changes:

    Map<Id, Origin> newOriginsMap = event.newOrigins.stream()
                .collect(toMap(Origin::id, o -> o));

        OriginChanges originChanges = new OriginChanges();

        concat(this.origins.keySet().stream(), newOriginsMap.keySet().stream()).collect(toSet()).forEach(
                originId -> {
                    Origin origin = newOriginsMap.get(originId);

                    if (isNewOrigin(originId, origin)) {
                        MonitoredOrigin monitoredOrigin = addMonitoredEndpoint(origin);
                        originChanges.addOrReplaceOrigin(originId, monitoredOrigin);

                    } else if (isUpdatedOrigin(originId, origin)) {
                        MonitoredOrigin monitoredOrigin = changeMonitoredEndpoint(origin);
                        originChanges.addOrReplaceOrigin(originId, monitoredOrigin);

                    } else if (isUnchangedOrigin(originId, origin)) {
                        LOG.info("Existing origin has been left unchanged. Origin={}:{}", appId, origin);
                        originChanges.keepExistingOrigin(originId, this.origins.get(originId));

                    } else if (isRemovedOrigin(originId, origin)) {
                        removeMonitoredEndpoint(originId);
                        originChanges.noteRemovedOrigin();
                    }
                }
        );

        this.origins = originChanges.updatedOrigins();

        if (originChanges.changed()) {
            notifyStateChange();
        }
    }

    private class OriginChanges {
        ImmutableMap.Builder<Id, MonitoredOrigin> monitoredOrigins = ImmutableMap.builder();
        AtomicBoolean changed = new AtomicBoolean(false);

        void addOrReplaceOrigin(Id originId, MonitoredOrigin origin) {
            monitoredOrigins.put(originId, origin);
            changed.set(true);
        }

        void keepExistingOrigin(Id originId, MonitoredOrigin origin) {
            monitoredOrigins.put(originId, origin);
        }

        void noteRemovedOrigin() {
            changed.set(true);
        }

        boolean changed() {
            return changed.get();
        }

        Map<Id, MonitoredOrigin> updatedOrigins() {
            return monitoredOrigins.build();
        }
    }

[mikkokar]

ok

[kvosper]

What is this test trying to prove? Its name doesn't explain.

[mikkokar]

ok

[kvosper]

This name is also not as informative as it could be. Perhaps something like "replaces origin when port number has changed"?

[mikkokar]

Which is a longwinded way of saying "(origins inventory) updates origin port number" ;-)

[kvosper]

This name is also not as informative as it could be. Perhaps something like "replaces origin when host name has changed"?

[mikkokar]

... again which is just a longer way saying "(origins inventory) updates origin host name".

[kvosper]

Stops and restarts?