/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 services;

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.util.Iterator;
import org.osoa.sca.annotations.Property;
import org.osoa.sca.annotations.Reference;

public class JobConfImpl implements JobConf {

	@Property
	public String input;
	@Property
	public String output;
	@Reference
	public Mapper myMapper;
	@Reference
	public IntermediateCollector iCollector;
	@Reference
	public OutputCollector oCollector;
	
	public void run() {
		
		FileInputStream fstream = null;
		try {
			fstream = new FileInputStream(input);
		}
		catch(FileNotFoundException e) {
			System.out.println("ERROR: Input file does not exist.");
			return;
		}
		
		DataInputStream in = new DataInputStream(fstream);
		BufferedReader br = new BufferedReader(new InputStreamReader(in));
	       
		//Read File Line By Line
	    String strLine = null;
	    int count = 1;
	    try {
	    	strLine = br.readLine();
	    }
	    catch(IOException e) {
	    	System.out.println("ERROR: Input file can not be read.");
	    	return;
	    }
	    while (strLine != null) {
	    	myMapper.map(count, strLine);
	    	count++;
	    	
	    	try {
		    	strLine = br.readLine();
		    }
		    catch(IOException e) {
		    	System.out.println("ERROR: Input file can not be read.");
		    	return;
		    }
	    }
	    
	    iCollector.reducePhase();

	    Iterator<KeyValuePair> it = oCollector.getCollection();
	    
	    //Write to output file
	    
	    FileOutputStream oStream = null;
	    try {
	    	oStream = new FileOutputStream(output);
	    }
	    catch(FileNotFoundException e) {
	    	System.out.println("ERROR: Output file can not be written to.");
	    	return;
	    }
	    
	    PrintWriter pw = new PrintWriter(oStream);
	    
	    while(it.hasNext()) {
	    	KeyValuePair kv = it.next();
	    	pw.println(kv.key + " " + kv.value);
	    }
	    
	    pw.close();
	}

}