// Path: bootique-job-consul/src/main/java/io/bootique/job/consul/lock/CompositeConsulLockHandler.java
// public class CompositeConsulLockHandler implements LockHandler {
// 
//     private LockHandler localLockHandler;
//     private LockHandler consulLockHandler;
// 
//     public CompositeConsulLockHandler(LockHandler localLockHandler, LockHandler consulLockHandler) {
//         this.localLockHandler = localLockHandler;
//         this.consulLockHandler = consulLockHandler;
//     }
// 
//     @Override
//     public RunnableJob lockingJob(RunnableJob executable, JobMetadata metadata) {
//         return localLockHandler.lockingJob(consulLockHandler.lockingJob(executable, metadata), metadata);
//     }
// }
// 
// Path: bootique-job-consul/src/main/java/io/bootique/job/consul/lock/ConsulLockHandler.java
// public class ConsulLockHandler implements LockHandler {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(ConsulLockHandler.class);
// 
//     private final KeyValueClient kvClient;
//     private final Supplier<String> consulSessionSupplier;
//     private final String serviceGroup;
// 
//     public ConsulLockHandler(KeyValueClient kvClient, Supplier<String> consulSessionSupplier, String serviceGroup) {
//         this.kvClient = kvClient;
//         this.consulSessionSupplier = consulSessionSupplier;
//         this.serviceGroup = serviceGroup;
//     }
// 
//     @Override
//     public RunnableJob lockingJob(RunnableJob executable, JobMetadata metadata) {
//         return () -> {
//             String lockName = getLockName(metadata);
// 
//             String sessionId = consulSessionSupplier.get();
// 
//             LOGGER.info("Attempting to lock '{}'", lockName);
//             boolean acquired = kvClient.acquireLock(lockName, sessionId);
//             if (!acquired) {
//                 LOGGER.info("** Another job instance owns the lock. Skipping execution of '{}'", lockName);
//                 return new JobResult(metadata, JobOutcome.SKIPPED, null,
//                         "Another job instance owns the lock. Skipping execution");
//             }
// 
//             try {
//                 return executable.run();
//             } finally {
//                 if (!kvClient.releaseLock(lockName, consulSessionSupplier.get())) {
//                     LOGGER.error("Failed to release lock, manual intervention might be needed: " + lockName);
//                 }
//             }
//         };
//     }
// 
//     private String getLockName(JobMetadata metadata) {
//         String jobName = metadata.getName();
//         return (serviceGroup == null || serviceGroup.isEmpty()) ?
//                 jobName : (serviceGroup + "/" + jobName);
//     }
// }
// 
// Path: bootique-job/src/main/java/io/bootique/job/lock/LocalLockHandler.java
// public class LocalLockHandler implements LockHandler {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(LocalLockHandler.class);
// 
//     private final ConcurrentMap<String, Lock> locks;
// 
//     public LocalLockHandler() {
//         this.locks = new ConcurrentHashMap<>();
//     }
// 
//     @Override
//     public RunnableJob lockingJob(RunnableJob executable, JobMetadata metadata) {
// 
//         return () -> {
//             String lockName = toLockName(metadata);
//             Lock lock = getLock(lockName);
// 
//             if (!lock.tryLock()) {
//                 LOGGER.info("Skipping execution of '{}', another job instance owns the lock.", metadata.getName());
//                 return new JobResult(
//                         metadata,
//                         JobOutcome.SKIPPED,
//                         null,
//                         "Skipping execution, another job instance owns the lock");
//             }
// 
//             LOGGER.info("Locked '{}'", metadata.getName());
// 
//             try {
//                 return executable.run();
//             } finally {
//                 lock.unlock();
//             }
//         };
//     }
// 
//     private Lock getLock(String lockName) {
//         return locks.computeIfAbsent(lockName, k -> new ReentrantLock());
//     }
// 
//     private String toLockName(JobMetadata metadata) {
//         return metadata.getName();
//     }
// }


import com.google.common.net.HostAndPort;
import com.orbitz.consul.Consul;
import io.bootique.annotation.BQConfig;
import io.bootique.annotation.BQConfigProperty;
import io.bootique.job.consul.lock.CompositeConsulLockHandler;
import io.bootique.job.consul.lock.ConsulLockHandler;
import io.bootique.job.lock.LocalLockHandler;
import io.bootique.job.lock.LockHandler;
import io.bootique.shutdown.ShutdownManager;
/*
 * Licensed to ObjectStyle LLC under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ObjectStyle LLC licenses
 * this file to you 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 io.bootique.job.consul;


/**
 * @since 3.0.M1
 */
@BQConfig("Consul jobs configuration")
public class ConsulLockHandlerFactory {

    private String consulHost;
    private Integer consulPort;
    private String dataCenter;
    private String serviceGroup;

    public ConsulLockHandlerFactory() {
        this.consulHost = "localhost";
        this.consulPort = 8500;
    }

    public CompositeConsulLockHandler createLockHandler(ShutdownManager shutdownManager) {

        String host = this.consulHost != null ? this.consulHost : "localhost";
        int port = this.consulPort != null ? this.consulPort : 8500;

        HostAndPort hostAndPort = HostAndPort.fromParts(host, port);
// Path: bootique-job-consul/src/main/java/io/bootique/job/consul/lock/CompositeConsulLockHandler.java
// public class CompositeConsulLockHandler implements LockHandler {
// 
//     private LockHandler localLockHandler;
//     private LockHandler consulLockHandler;
// 
//     public CompositeConsulLockHandler(LockHandler localLockHandler, LockHandler consulLockHandler) {
//         this.localLockHandler = localLockHandler;
//         this.consulLockHandler = consulLockHandler;
//     }
// 
//     @Override
//     public RunnableJob lockingJob(RunnableJob executable, JobMetadata metadata) {
//         return localLockHandler.lockingJob(consulLockHandler.lockingJob(executable, metadata), metadata);
//     }
// }
// 
// Path: bootique-job-consul/src/main/java/io/bootique/job/consul/lock/ConsulLockHandler.java
// public class ConsulLockHandler implements LockHandler {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(ConsulLockHandler.class);
// 
//     private final KeyValueClient kvClient;
//     private final Supplier<String> consulSessionSupplier;
//     private final String serviceGroup;
// 
//     public ConsulLockHandler(KeyValueClient kvClient, Supplier<String> consulSessionSupplier, String serviceGroup) {
//         this.kvClient = kvClient;
//         this.consulSessionSupplier = consulSessionSupplier;
//         this.serviceGroup = serviceGroup;
//     }
// 
//     @Override
//     public RunnableJob lockingJob(RunnableJob executable, JobMetadata metadata) {
//         return () -> {
//             String lockName = getLockName(metadata);
// 
//             String sessionId = consulSessionSupplier.get();
// 
//             LOGGER.info("Attempting to lock '{}'", lockName);
//             boolean acquired = kvClient.acquireLock(lockName, sessionId);
//             if (!acquired) {
//                 LOGGER.info("** Another job instance owns the lock. Skipping execution of '{}'", lockName);
//                 return new JobResult(metadata, JobOutcome.SKIPPED, null,
//                         "Another job instance owns the lock. Skipping execution");
//             }
// 
//             try {
//                 return executable.run();
//             } finally {
//                 if (!kvClient.releaseLock(lockName, consulSessionSupplier.get())) {
//                     LOGGER.error("Failed to release lock, manual intervention might be needed: " + lockName);
//                 }
//             }
//         };
//     }
// 
//     private String getLockName(JobMetadata metadata) {
//         String jobName = metadata.getName();
//         return (serviceGroup == null || serviceGroup.isEmpty()) ?
//                 jobName : (serviceGroup + "/" + jobName);
//     }
// }
// 
// Path: bootique-job/src/main/java/io/bootique/job/lock/LocalLockHandler.java
// public class LocalLockHandler implements LockHandler {
// 
//     private static final Logger LOGGER = LoggerFactory.getLogger(LocalLockHandler.class);
// 
//     private final ConcurrentMap<String, Lock> locks;
// 
//     public LocalLockHandler() {
//         this.locks = new ConcurrentHashMap<>();
//     }
// 
//     @Override
//     public RunnableJob lockingJob(RunnableJob executable, JobMetadata metadata) {
// 
//         return () -> {
//             String lockName = toLockName(metadata);
//             Lock lock = getLock(lockName);
// 
//             if (!lock.tryLock()) {
//                 LOGGER.info("Skipping execution of '{}', another job instance owns the lock.", metadata.getName());
//                 return new JobResult(
//                         metadata,
//                         JobOutcome.SKIPPED,
//                         null,
//                         "Skipping execution, another job instance owns the lock");
//             }
// 
//             LOGGER.info("Locked '{}'", metadata.getName());
// 
//             try {
//                 return executable.run();
//             } finally {
//                 lock.unlock();
//             }
//         };
//     }
// 
//     private Lock getLock(String lockName) {
//         return locks.computeIfAbsent(lockName, k -> new ReentrantLock());
//     }
// 
//     private String toLockName(JobMetadata metadata) {
//         return metadata.getName();
//     }
// }


// Path: bootique-job-consul/src/main/java/io/bootique/job/consul/ConsulLockHandlerFactory.java
import com.google.common.net.HostAndPort;
import com.orbitz.consul.Consul;
import io.bootique.annotation.BQConfig;
import io.bootique.annotation.BQConfigProperty;
import io.bootique.job.consul.lock.CompositeConsulLockHandler;
import io.bootique.job.consul.lock.ConsulLockHandler;
import io.bootique.job.lock.LocalLockHandler;
import io.bootique.job.lock.LockHandler;
import io.bootique.shutdown.ShutdownManager;

/*
 * Licensed to ObjectStyle LLC under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ObjectStyle LLC licenses
 * this file to you 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 io.bootique.job.consul;


/**
 * @since 3.0.M1
 */
@BQConfig("Consul jobs configuration")
public class ConsulLockHandlerFactory {

    private String consulHost;
    private Integer consulPort;
    private String dataCenter;
    private String serviceGroup;

    public ConsulLockHandlerFactory() {
        this.consulHost = "localhost";
        this.consulPort = 8500;
    }

    public CompositeConsulLockHandler createLockHandler(ShutdownManager shutdownManager) {

        String host = this.consulHost != null ? this.consulHost : "localhost";
        int port = this.consulPort != null ? this.consulPort : 8500;

        HostAndPort hostAndPort = HostAndPort.fromParts(host, port);
        Consul consul = Consul.builder().withHostAndPort(hostAndPort).build();