import javax.inject.Inject;
@Import(value = {LocalDbAuthConfig.class,CassandraAvailabiltyConfig.class})
@Configuration
@EnableRetry
@EnableAspectJAutoProxy
@EnableDbAuthPropertySupport
public class CassandraConfig  extends AbstractCassandraConfiguration {

  private Logger LOGGER= LogManager.getLogger();

  @Value("${cassandra.contact.hosts}")
  private String cassandraContactPoints;

  @Value("${cassandra.contact.port}")
  private Integer cassandraPort;

  @Value("${cassandra.auth.alias}")
  private String cassandraDbAuthAlias;

  private String cassandraUsername;

  private String cassandraPassword;

  @Value("${cassandra.schema.create}")
  private boolean createSchema;

  @Inject
  private DbAuthBean dbAuthBean;

  @Inject
  private RetryTemplate retryTemplate;

  @Inject
  private ApplicationContext context;


  @Bean
  @DependsOn("assertCassandraAvailable")
  public CassandraClusterFactoryBean cluster() {

    cassandraUsername = dbAuthBean.property(cassandraDbAuthAlias, "username");
    cassandraPassword = dbAuthBean.property(cassandraDbAuthAlias, "password");

    MyCassandraClusterFactoryBean cluster = new MyCassandraClusterFactoryBean();
    cluster.setContactPoints(cassandraContactPoints);
    cluster.setPort(cassandraPort);
    cluster.setUsername(cassandraUsername);
    cluster.setPassword(cassandraPassword);
    cluster.setQueryOptions(new QueryOptions().setConsistencyLevel(ConsistencyLevel.ALL));
    cluster.setReconnectionPolicy(new ConstantReconnectionPolicy(100L));

    return cluster;
  }

  @Bean
  @DependsOn("assertCassandraAvailable")
  public CassandraSessionFactoryBean session() throws Exception {

    MyCassandraSessionFactoryBean session = new MyCassandraSessionFactoryBean();
    session.setCluster(cluster().getObject());
    session.setConverter(cassandraConverter());
    session.setKeyspaceName(getKeyspaceName());
    session.setSchemaAction(getSchemaAction());
    session.setStartupScripts(getStartupScripts());
    session.setShutdownScripts(getShutdownScripts());
    return session;
  }

  @Bean
  public RetryTemplate retryTemplate() {
    RetryTemplate retryTemplate = new RetryTemplate();

    FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
    fixedBackOffPolicy.setBackOffPeriod(200l);
    retryTemplate.setBackOffPolicy(fixedBackOffPolicy);

    SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
    retryPolicy.setMaxAttempts(5);
    retryTemplate.setRetryPolicy(retryPolicy);

    return retryTemplate;
  }

  /**
   *Extend the Bean so that we can override it's Singleton behaviour.
   */
  public class MyCassandraClusterFactoryBean extends CassandraClusterFactoryBean
  {
    @Override public boolean isSingleton() {
      return false;
    }

    @Override public Cluster getObject() {
      return super.getObject();
    }

    @Override public void afterPropertiesSet() throws Exception {
      super.afterPropertiesSet();
    }
  }
  public class MyCassandraSessionFactoryBean extends CassandraSessionFactoryBean {
    public MyCassandraSessionFactoryBean() {
      super();
    }

    @Override public void afterPropertiesSet() throws Exception {

      retryTemplate.execute(new RetryCallback<Void, Exception>() {
        @Override public Void doWithRetry(RetryContext retryContext) throws Exception {
          try {
            MyCassandraSessionFactoryBean.super.afterPropertiesSet();
          }
          catch (Exception e) {
            cluster().afterPropertiesSet();
            Cluster cluster = cluster().getObject();
            setCluster(cluster);
            throw e;
          }
          return null;
        }
      });

    }
  }

  @Override
  public String[] getEntityBasePackages() {
      return new String[]{Component.class.getPackage().getName()};
  }

  @Value("${cassandra.keyspace}")
  private String cassandraKeyspace;

  @Override
  protected String getKeyspaceName() {
    return cassandraKeyspace;
  }

  @Override
  public SchemaAction getSchemaAction() {
    if (createSchema) {
      return SchemaAction.CREATE_IF_NOT_EXISTS;
    }
    else {
      return SchemaAction.NONE;
    }
  }

  @Bean
  public CassandraMappingContext mappingContext() {
    return new BasicCassandraMappingContext();
  }

  @Bean
  public CassandraConverter converter() {
    return new MappingCassandraConverter(mappingContext());
  }

  @Override
  public RetryPolicy getRetryPolicy() {
    return DefaultRetryPolicy.INSTANCE;
  }
}