/*
Copyright [2017] [DCS <Info-dcs@diehl.com>]

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
 */

package org.diehl.dcs.kalinka.activemq_plugin;


import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter;
import org.apache.activemq.broker.BrokerPlugin;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ConnectionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author michas <michas@jarmoni.org>
 *
 */
public class ActivemqPlugin implements BrokerPlugin {

	private static final Logger LOG = LoggerFactory.getLogger(ActivemqPlugin.class);

	private final ZkClient zkClient;
	private final String host;

	public ActivemqPlugin(final String zkServers, final String host) {

		LOG.debug("Trying to create ZkClient for zkServers=" + zkServers + ", currentHost=" + host);
		this.zkClient = new ZkClient(zkServers);
		this.host = host;
		LOG.info("Created ZkClient for zkServers=" + zkServers + ", currentHost=" + host);
	}


	@Override
	public Broker installPlugin(final Broker broker) throws Exception {

		return new BrokerFilter(broker) {

			@Override
			public void addConnection(final ConnectionContext context, final ConnectionInfo info)
					throws Exception {

				super.addConnection(context, info);
				upsertZkNode(context.getClientId());
			}

			@Override
			public void removeConnection(final ConnectionContext context, final ConnectionInfo info,
				final Throwable error) throws Exception {

				super.removeConnection(context, info, error);
				deleteZkNode(context.getClientId());
			}
		};
	};

	void upsertZkNode(final String clientId) {

		final String node = "/" + clientId;
		LOG.debug("Trying to upsert node={}, to host={}", node, this.host);
		try {
			this.zkClient.createPersistent(node, this.host);
		}
		catch(final ZkNodeExistsException e) {
			this.deleteZkNode(clientId);
			this.zkClient.createPersistent(node, this.host);
		}
		LOG.info("Upserted node={}, to host={}", node, this.host);
	}

	void deleteZkNode(final String clientId) {

		final String node = "/" + clientId;
		LOG.debug("Trying to delete node={}", node);
		this.zkClient.delete(node);
		LOG.info("Deleted node={}", node);
	}

}

@@ -1,11 +1,14 @@
 /*%0A
+ * 
 Copyrigh
@@ -42,18 +42,17 @@
 hl.com%3E%5D
-%0A%0A
+ 
 Licensed
@@ -94,16 +94,19 @@
 2.0 (the
+%0A *
  %22Licens
@@ -109,17 +109,17 @@
 cense%22);
-%0A
+ 
 you may 
@@ -174,17 +174,17 @@
 License.
-%0A
+ 
 You may 
@@ -191,16 +191,19 @@
 obtain a
+%0A *
  copy of
@@ -221,13 +221,8 @@
 e at
-%0A%0A   
  htt
@@ -260,18 +260,17 @@
 ENSE-2.0
-%0A%0A
+ 
 Unless r
@@ -290,16 +290,19 @@
 plicable
+%0A *
  law or 
@@ -331,17 +331,17 @@
 software
-%0A
+ 
 distribu
@@ -394,16 +394,19 @@
  IS%22
+%0A *
  BASIS,
-%0A
+ 
 WITH
@@ -473,17 +473,17 @@
 implied.
-%0A
+ 
 See the 
@@ -489,16 +489,19 @@
  License
+%0A *
  for the
@@ -544,17 +544,17 @@
 ions and
-%0A
+ 
 limitati
@@ -1215,16 +1215,87 @@
 lass);%0A%0A
+%09private static final String CLIENT_ID_PREFIX_KAFKA = %22kafka-client%22;%0A%0A
 %09private
@@ -1475,42 +1475,39 @@
 ers=
-%22 + zkServers + %22, currentHost=%22 +
+%7B%7D, currentHost=%7B%7D%22, zkServers,
  hos
@@ -1916,43 +1916,206 @@
 %09%09%09%09
-super.addConnection(context, info);
+LOG.debug(%22Received connect from clientId=%7B%7D%22, context.getClientId());%0A%0A%09%09%09%09super.addConnection(context, info);%0A%09%09%09%09if (context.getClientId().startsWith(CLIENT_ID_PREFIX_KAFKA)) %7B%0A%09%09%09%09%09return;%0A%09%09%09%09%7D
 %0A%09%09%09
@@ -2267,16 +2267,17 @@
 fo,%0A%09%09%09%09
+%09
 final Th
@@ -2319,53 +2319,219 @@
 %09%09%09%09
-super.removeConnection(context, info, error);
+LOG.debug(%22Received disconnect from clientId=%7B%7D%22, context.getClientId());%0A%0A%09%09%09%09super.removeConnection(context, info, error);%0A%09%09%09%09if (context.getClientId().startsWith(CLIENT_ID_PREFIX_KAFKA)) %7B%0A%09%09%09%09%09return;%0A%09%09%09%09%7D
 %0A%09%09%09
@@ -2803,16 +2803,15 @@
 %0A%09%09%7D
-%0A%09%09
+ 
 catch
+ 
 (fin
@@ -3193,11 +3193,10 @@
 de);%0A%09%7D%0A
-%0A
 %7D%0A

check clientId in activemq-plugin