// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.documentapi.messagebus.protocol;

import com.yahoo.config.subscription.ConfigSourceSet;
import com.yahoo.config.subscription.ConfigSubscriber;
import com.yahoo.jrt.Supervisor;
import com.yahoo.jrt.Transport;
import com.yahoo.jrt.slobrok.api.IMirror;
import com.yahoo.jrt.slobrok.api.Mirror;
import com.yahoo.jrt.slobrok.api.SlobrokList;
import com.yahoo.messagebus.routing.RoutingContext;
import com.yahoo.cloud.config.SlobroksConfig;

import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Abstract class for policies that allow you to specify which slobrok to use for the
 * routing.
 */
public abstract class ExternalSlobrokPolicy extends AsyncInitializationPolicy implements ConfigSubscriber.SingleSubscriber<SlobroksConfig> {
    String error;
    private Supervisor orb = null;
    private final AtomicReference<Mirror> safeMirror = new AtomicReference<>(null);
    private SlobrokList slobroks = null;
    private final AtomicBoolean firstTry = new AtomicBoolean(true);
    private ConfigSubscriber subscriber;
    String[] configSources = null;
    private final static String slobrokConfigId = "admin/slobrok.0";


    ExternalSlobrokPolicy(Map<String, String> param) {
        super();

        String conf = param.get("config");
        if (conf != null) {
            configSources = conf.split(",");
        }

        String slbrk = param.get("slobroks");
        if (slbrk != null) {
            slobroks = new SlobrokList();
            slobroks.setup(slbrk.split(","));
        }

        if (slobroks != null || configSources != null) {
            needAsynchronousInitialization();
        }
    }

    @Override
    public synchronized void init() {
        if (slobroks != null) {
            orb = new Supervisor(new Transport());
            safeMirror.set(new Mirror(orb, slobroks));
        }

        if (configSources != null) {
            if (safeMirror.get() == null) {
                orb = new Supervisor(new Transport());
                subscriber = subscribe(slobrokConfigId, new ConfigSourceSet(configSources));
            }
        }
    }

    private ConfigSubscriber subscribe(String configId, final ConfigSourceSet configSourceSet) {
        ConfigSubscriber subscriber = new ConfigSubscriber(configSourceSet);
        subscriber.subscribe(this, SlobroksConfig.class, configId);
        return subscriber;
    }

    public IMirror getMirror() {
        return safeMirror.get();
    }

    public  List<Mirror.Entry> lookup(RoutingContext context, String pattern) {
        IMirror mirror1 = getMirror();
        if (mirror1 == null) {
            mirror1 = context.getMirror();
        }

        List<Mirror.Entry> arr = mirror1.lookup(pattern);

        if ((arr.isEmpty()) && firstTry.get()) {
            synchronized(this)  {
                try {
                    int count = 0;
                    while (arr.isEmpty() && count < 100) {
                        Thread.sleep(50);
                        arr = mirror1.lookup(pattern);
                        count++;
                    }
                } catch (InterruptedException e) {
                }

            }
        }

        firstTry.set(false);
        return arr;
    }

    @Override
    public synchronized void configure(SlobroksConfig config) {
        String[] slist = new String[config.slobrok().size()];

        for(int i = 0; i < config.slobrok().size(); i++) {
            slist[i] = config.slobrok(i).connectionspec();
        }
        if (slobroks == null) {
            slobroks = new SlobrokList();
        }
        slobroks.setup(slist);
        if (safeMirror.get() == null) {
            safeMirror.set(new Mirror(orb, slobroks));
        }

    }

    @Override
    public synchronized void destroy() {
        if (subscriber!=null) subscriber.close();
    }

}

@@ -1155,21 +1155,9 @@
 ate 
-final AtomicB
+b
 oole
@@ -1174,31 +1174,12 @@
 y = 
-new AtomicBoolean(
 true
-)
 ;%0A  
@@ -2901,17 +2901,16 @@
      if 
-(
 (arr.isE
@@ -2920,27 +2920,8 @@
 y())
- && firstTry.get())
  %7B%0A 
@@ -2973,14 +2973,54 @@
     
-t
+if (firstT
 ry
+)
  %7B%0A
+                    try %7B%0A    
     
@@ -3050,16 +3050,20 @@
 nt = 0;%0A
+    
         
@@ -3137,16 +3137,20 @@
         
+    
 Thread.s
@@ -3159,16 +3159,20 @@
 ep(50);%0A
+    
         
@@ -3242,16 +3242,20 @@
         
+    
 count++;
@@ -3267,34 +3267,42 @@
                 
-%7D%0A
+    %7D%0A    
                 
@@ -3332,24 +3332,83 @@
 eption e) %7B%0A
+                    %7D%0A                    firstTry = true;%0A
             
@@ -3443,37 +3443,8 @@
  %7D%0A%0A
-        firstTry.set(false);%0A
     

Use syncronized and ordinary boolean firstTry