/*
 * 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.swarm.tokenization.common;

import com.google.api.services.bigquery.model.TableRow;
import com.google.common.collect.ImmutableMap;
import org.apache.beam.sdk.io.gcp.pubsub.PubsubMessage;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.values.KV;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("serial")
public class PubSubMessageConverts extends DoFn<KV<String, TableRow>, PubsubMessage> {
  private static final Logger LOG = LoggerFactory.getLogger(PubSubMessageConverts.class);

  @ProcessElement
  public void processContext(ProcessContext c) {
    String tableRef = c.element().getKey();
    c.element()
        .getValue()
        .getF()
        .iterator()
        .forEachRemaining(
            value -> {
              String json = Util.gson.toJson(value);
              LOG.info("Json {}", json);
              PubsubMessage message =
                  new PubsubMessage(
                      json.getBytes(),
                      ImmutableMap.<String, String>builder().put("table_name", tableRef).build());
              c.output(message);
            });
  }
}

@@ -1209,17 +1209,16 @@
 c) %7B%0A   
- 
 String t
@@ -1251,16 +1251,31 @@
 ey();%0A  
+ TableRow row =
   c.elem
@@ -1279,25 +1279,16 @@
 lement()
-%0A        
 .getValu
@@ -1294,106 +1294,10 @@
 ue()
+;
 %0A
-        .getF()%0A        .iterator()%0A        .forEachRemaining(%0A            value -%3E %7B%0A           
    S
@@ -1330,27 +1330,21 @@
 son(
-value);%0A           
+row.getF());%0A
    L
@@ -1369,28 +1369,17 @@
  json);%0A
-            
+%09
   Pubsub
@@ -1396,34 +1396,23 @@
 ssage =%0A
-       
+%09
       
-     
 new Pubs
@@ -1422,28 +1422,17 @@
 essage(%0A
-            
+%09
         
@@ -1450,25 +1450,17 @@
 ytes(),%0A
-         
+%09
         
@@ -1461,19 +1461,16 @@
         
-   
 Immutabl
@@ -1538,28 +1538,17 @@
 ild());%0A
-            
+%09
   c.outp
@@ -1568,19 +1568,8 @@
     
-        %7D);
 %0A  %7D

pardo converts from simple func