// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/PslCredentialsProvider.java
// public class PslCredentialsProvider implements CredentialsProvider {
// 
//   private final Credentials credentials;
// 
//   public PslCredentialsProvider(@Nullable String credentialsKey) {
//     this.credentials =
//         credentialsKey != null
//             ? createCredentialsFromKey(credentialsKey)
//             : createDefaultCredentials();
//   }
// 
//   private static Credentials createCredentialsFromKey(String key) {
//     try {
//       return GoogleCredentials.fromStream(new ByteArrayInputStream(Base64.decodeBase64(key)))
//           .createScoped("https://www.googleapis.com/auth/cloud-platform");
//     } catch (IOException e) {
//       throw new UncheckedIOException("Failed to create Credentials from key", e);
//     }
//   }
// 
//   public static Credentials createDefaultCredentials() {
//     try {
//       return GoogleCredentials.getApplicationDefault()
//           .createScoped("https://www.googleapis.com/auth/cloud-platform");
//     } catch (IOException e) {
//       throw new UncheckedIOException("Failed to create default Credentials", e);
//     }
//   }
// 
//   @Override
//   public Credentials getCredentials() {
//     return credentials;
//   }
// }


import static com.google.cloud.pubsublite.internal.ExtractStatus.toCanonical;
import static com.google.cloud.pubsublite.internal.wire.ServiceClients.addDefaultSettings;
import static com.google.cloud.pubsublite.internal.wire.ServiceClients.getCallContext;
import com.google.api.gax.rpc.ApiCallContext;
import com.google.api.gax.rpc.ApiException;
import com.google.auto.value.AutoValue;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.MessageMetadata;
import com.google.cloud.pubsublite.Partition;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings;
import com.google.cloud.pubsublite.internal.Publisher;
import com.google.cloud.pubsublite.internal.wire.PartitionCountWatchingPublisherSettings;
import com.google.cloud.pubsublite.internal.wire.PartitionPublisherFactory;
import com.google.cloud.pubsublite.internal.wire.PubsubContext;
import com.google.cloud.pubsublite.internal.wire.RoutingMetadata;
import com.google.cloud.pubsublite.internal.wire.SinglePartitionPublisherBuilder;
import com.google.cloud.pubsublite.spark.internal.PslCredentialsProvider;
import com.google.cloud.pubsublite.v1.AdminServiceClient;
import com.google.cloud.pubsublite.v1.AdminServiceSettings;
import com.google.cloud.pubsublite.v1.PublisherServiceClient;
import com.google.cloud.pubsublite.v1.PublisherServiceSettings;
import java.io.Serializable;
import javax.annotation.Nullable;
import org.apache.spark.sql.sources.v2.DataSourceOptions;
/*
 * Copyright 2020 Google LLC
 *
 * 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 com.google.cloud.pubsublite.spark;



@AutoValue
public abstract class PslWriteDataSourceOptions implements Serializable {

  @Nullable
  public abstract String credentialsKey();

  public abstract TopicPath topicPath();

  public static Builder builder() {
    return new AutoValue_PslWriteDataSourceOptions.Builder().setCredentialsKey(null);
  }

  @AutoValue.Builder
  public abstract static class Builder {

    public abstract PslWriteDataSourceOptions.Builder setCredentialsKey(String credentialsKey);

    public abstract PslWriteDataSourceOptions.Builder setTopicPath(TopicPath topicPath);

    public abstract PslWriteDataSourceOptions build();
  }

  public static PslWriteDataSourceOptions fromSparkDataSourceOptions(DataSourceOptions options) {
    if (!options.get(Constants.TOPIC_CONFIG_KEY).isPresent()) {
      throw new IllegalArgumentException(Constants.TOPIC_CONFIG_KEY + " is required.");
    }

    Builder builder = builder();
    String topicPathVal = options.get(Constants.TOPIC_CONFIG_KEY).get();
    try {
      builder.setTopicPath(TopicPath.parse(topicPathVal));
    } catch (ApiException e) {
      throw new IllegalArgumentException("Unable to parse topic path " + topicPathVal, e);
    }
    options.get(Constants.CREDENTIALS_KEY_CONFIG_KEY).ifPresent(builder::setCredentialsKey);
    return builder.build();
  }

  public PslCredentialsProvider getCredentialProvider() {
// Path: src/main/java/com/google/cloud/pubsublite/spark/internal/PslCredentialsProvider.java
// public class PslCredentialsProvider implements CredentialsProvider {
// 
//   private final Credentials credentials;
// 
//   public PslCredentialsProvider(@Nullable String credentialsKey) {
//     this.credentials =
//         credentialsKey != null
//             ? createCredentialsFromKey(credentialsKey)
//             : createDefaultCredentials();
//   }
// 
//   private static Credentials createCredentialsFromKey(String key) {
//     try {
//       return GoogleCredentials.fromStream(new ByteArrayInputStream(Base64.decodeBase64(key)))
//           .createScoped("https://www.googleapis.com/auth/cloud-platform");
//     } catch (IOException e) {
//       throw new UncheckedIOException("Failed to create Credentials from key", e);
//     }
//   }
// 
//   public static Credentials createDefaultCredentials() {
//     try {
//       return GoogleCredentials.getApplicationDefault()
//           .createScoped("https://www.googleapis.com/auth/cloud-platform");
//     } catch (IOException e) {
//       throw new UncheckedIOException("Failed to create default Credentials", e);
//     }
//   }
// 
//   @Override
//   public Credentials getCredentials() {
//     return credentials;
//   }
// }


// Path: src/main/java/com/google/cloud/pubsublite/spark/PslWriteDataSourceOptions.java
import static com.google.cloud.pubsublite.internal.ExtractStatus.toCanonical;
import static com.google.cloud.pubsublite.internal.wire.ServiceClients.addDefaultSettings;
import static com.google.cloud.pubsublite.internal.wire.ServiceClients.getCallContext;
import com.google.api.gax.rpc.ApiCallContext;
import com.google.api.gax.rpc.ApiException;
import com.google.auto.value.AutoValue;
import com.google.cloud.pubsublite.AdminClient;
import com.google.cloud.pubsublite.AdminClientSettings;
import com.google.cloud.pubsublite.MessageMetadata;
import com.google.cloud.pubsublite.Partition;
import com.google.cloud.pubsublite.TopicPath;
import com.google.cloud.pubsublite.cloudpubsub.PublisherSettings;
import com.google.cloud.pubsublite.internal.Publisher;
import com.google.cloud.pubsublite.internal.wire.PartitionCountWatchingPublisherSettings;
import com.google.cloud.pubsublite.internal.wire.PartitionPublisherFactory;
import com.google.cloud.pubsublite.internal.wire.PubsubContext;
import com.google.cloud.pubsublite.internal.wire.RoutingMetadata;
import com.google.cloud.pubsublite.internal.wire.SinglePartitionPublisherBuilder;
import com.google.cloud.pubsublite.spark.internal.PslCredentialsProvider;
import com.google.cloud.pubsublite.v1.AdminServiceClient;
import com.google.cloud.pubsublite.v1.AdminServiceSettings;
import com.google.cloud.pubsublite.v1.PublisherServiceClient;
import com.google.cloud.pubsublite.v1.PublisherServiceSettings;
import java.io.Serializable;
import javax.annotation.Nullable;
import org.apache.spark.sql.sources.v2.DataSourceOptions;

/*
 * Copyright 2020 Google LLC
 *
 * 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 com.google.cloud.pubsublite.spark;



@AutoValue
public abstract class PslWriteDataSourceOptions implements Serializable {

  @Nullable
  public abstract String credentialsKey();

  public abstract TopicPath topicPath();

  public static Builder builder() {
    return new AutoValue_PslWriteDataSourceOptions.Builder().setCredentialsKey(null);
  }

  @AutoValue.Builder
  public abstract static class Builder {

    public abstract PslWriteDataSourceOptions.Builder setCredentialsKey(String credentialsKey);

    public abstract PslWriteDataSourceOptions.Builder setTopicPath(TopicPath topicPath);

    public abstract PslWriteDataSourceOptions build();
  }

  public static PslWriteDataSourceOptions fromSparkDataSourceOptions(DataSourceOptions options) {
    if (!options.get(Constants.TOPIC_CONFIG_KEY).isPresent()) {
      throw new IllegalArgumentException(Constants.TOPIC_CONFIG_KEY + " is required.");
    }

    Builder builder = builder();
    String topicPathVal = options.get(Constants.TOPIC_CONFIG_KEY).get();
    try {
      builder.setTopicPath(TopicPath.parse(topicPathVal));
    } catch (ApiException e) {
      throw new IllegalArgumentException("Unable to parse topic path " + topicPathVal, e);
    }
    options.get(Constants.CREDENTIALS_KEY_CONFIG_KEY).ifPresent(builder::setCredentialsKey);
    return builder.build();
  }

  public PslCredentialsProvider getCredentialProvider() {
    return new PslCredentialsProvider(credentialsKey());