/*
 * Copyright © 2015 Cask Data, Inc.
 *
 * 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 co.cask.gcs.sink.io;

import com.google.api.gax.core.FixedCredentialsProvider;
import com.google.auth.Credentials;
import com.google.auth.oauth2.ServiceAccountCredentials;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import javax.annotation.Nullable;

/**
 * Writer that writes a single record
 */
public class PubSubRecordWriter extends RecordWriter<NullWritable, Text> {

  private static final Logger LOG = LoggerFactory.getLogger(PubSubRecordWriter.class);
	private  Publisher publisher;

	public PubSubRecordWriter(final String projectId, final String topicId,
														final String credentialsPath) throws IOException {

		TopicName topic = TopicName.create(projectId, topicId);
		this.publisher = Publisher.defaultBuilder(topic)
      .setCredentialsProvider(new FixedCredentialsProvider() {
        @Nullable
        @Override
        public Credentials getCredentials() {
          return loadCredentials(credentialsPath);
        }
      }).build();
	}

	private Credentials loadCredentials (String credentialsPath) {
		File path = new File(credentialsPath);
		if (!path.exists()) {
			throw new IllegalArgumentException("credentialsPath does not exist");
		}

		try (FileInputStream serviceAccountStream = new FileInputStream(credentialsPath)) {
				return ServiceAccountCredentials.fromStream(serviceAccountStream);
    } catch (IOException e) {
      throw new IllegalArgumentException(String.format("Exception reading from credentials file %s",
                                                       credentialsPath));
    }
	}

	@Override
	public void write(NullWritable key, Text value) throws IOException {
		if (publisher == null) {
			throw new IOException("PubSub publisher is not initialized");
		}
		ByteString data = ByteString.copyFromUtf8(value.toString());
		PubsubMessage message = PubsubMessage.newBuilder().setData(data).build();

		try {
      publisher.publish(message).get();
    } catch (InterruptedException | ExecutionException e) {
      throw new IOException(e);
    }
  }

	@Override
	public void close(TaskAttemptContext taskAttemptContext) {
		try {
			publisher.shutdown();
		} catch (Exception e) {
			LOG.warn("Error while shutting down publisher", e.getMessage());
  	}
	}
}

@@ -1548,25 +1548,26 @@
 class);%0A
-%09
+  
 private 
  Publish
@@ -1558,17 +1558,16 @@
 private 
- 
 Publishe
@@ -1580,17 +1580,18 @@
 isher;%0A%0A
-%09
+  
 public P
@@ -1658,22 +1658,36 @@
 Id,%0A
-%09%09%09%09%09%09%09%09%09%09%09%09%09%09
+                            
 fina
@@ -1734,18 +1734,20 @@
 tion %7B%0A%0A
-%09%09
+    
 TopicNam
@@ -1794,18 +1794,20 @@
 picId);%0A
-%09%09
+    
 this.pub
@@ -2071,21 +2071,23 @@
 uild();%0A
-%09%7D%0A%0A%09
+  %7D%0A%0A  
 private 
@@ -2113,17 +2113,16 @@
 dentials
- 
 (String 
@@ -2140,18 +2140,20 @@
 Path) %7B%0A
-%09%09
+    
 File pat
@@ -2183,18 +2183,20 @@
 sPath);%0A
-%09%09
+    
 if (!pat
@@ -2205,27 +2205,30 @@
 exists()) %7B%0A
-%09%09%09
+      
 throw new Il
@@ -2289,15 +2289,19 @@
 %22);%0A
-%09%09%7D%0A%0A%09%09
+    %7D%0A%0A    
 try 
@@ -2384,12 +2384,14 @@
 ) %7B%0A
-%09%09%09%09
+      
 retu
@@ -2668,13 +2668,15 @@
   %7D%0A
-%09%7D%0A%0A%09
+  %7D%0A%0A  
 @Ove
@@ -2673,33 +2673,34 @@
  %7D%0A%0A  @Override%0A
-%09
+  
 public void writ
@@ -2752,18 +2752,20 @@
 ption %7B%0A
-%09%09
+    
 if (publ
@@ -2781,19 +2781,22 @@
 null) %7B%0A
-%09%09%09
+      
 throw ne
@@ -2853,14 +2853,18 @@
 %22);%0A
-%09%09%7D%0A%09%09
+    %7D%0A    
 Byte
@@ -2920,18 +2920,20 @@
 ing());%0A
-%09%09
+    
 PubsubMe
@@ -2999,18 +2999,20 @@
 ild();%0A%0A
-%09%09
+    
 try %7B%0A  
@@ -3152,17 +3152,18 @@
  %7D%0A  %7D%0A%0A
-%09
+  
 @Overrid
@@ -3164,17 +3164,18 @@
 verride%0A
-%09
+  
 public v
@@ -3229,19 +3229,24 @@
 ) %7B%0A
-%09%09
+    
 try %7B%0A
-%09%09%09
+      
 publ
@@ -3263,18 +3263,20 @@
 down();%0A
-%09%09
+    
 %7D catch 
@@ -3295,11 +3295,14 @@
 ) %7B%0A
-%09%09%09
+      
 LOG.
@@ -3368,12 +3368,14 @@
 ;%0A  
-%09%7D%0A%09
+  %7D%0A  
 %7D%0A%7D%0A

Reformat code