/*
 * Copyright 2013 the original author or authors.
 *
 * 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 org.elasticsearch.hadoop.serialization;

import org.elasticsearch.hadoop.cfg.ConfigurationOptions;
import org.elasticsearch.hadoop.cfg.Settings;
import org.elasticsearch.hadoop.util.BytesArray;
import org.elasticsearch.hadoop.util.StringUtils;

public class UpdateCommand extends AbstractCommand {

    private final boolean UPSERT_DOC;

    private final byte[] HEADER_PREFIX = ("{\"" + ConfigurationOptions.ES_OPERATION_UPDATE + "\":{\"_id\":\"").getBytes(StringUtils.UTF_8);
    private final byte[] HEADER_SUFFIX = ("\"}}\n{\"doc\":").getBytes(StringUtils.UTF_8);
    private final byte[] BODY_NO_DOC_UPSERT_SUFFIX = ("}").getBytes(StringUtils.UTF_8);
    private final byte[] BODY_DOC_UPSERT_SUFFIX = (",\"doc_as_upsert\":true}").getBytes(StringUtils.UTF_8);


    UpdateCommand(Settings settings) {
        super(settings);
        UPSERT_DOC = settings.getUpsertDoc();
    }

    @Override
    protected boolean isIdRequired() {
        return true;
    }

    @Override
    protected byte[] headerPrefix() {
        return HEADER_PREFIX;
    }

    @Override
    protected byte[] headerSuffix() {
        return HEADER_SUFFIX;
    }

    @Override
    protected byte[] header() {
        throw new UnsupportedOperationException("Id is required but none was given");
    }


    @Override
    public int prepare(Object object) {
        return super.prepare(object) + (UPSERT_DOC ? BODY_DOC_UPSERT_SUFFIX.length : BODY_NO_DOC_UPSERT_SUFFIX.length);
    }

    @Override
    protected void writeSource(Object object, BytesArray buffer) {
        // write document
        super.writeSource(object, buffer);
        // append body
        if (UPSERT_DOC) {
            copyIntoBuffer(BODY_DOC_UPSERT_SUFFIX, buffer);
        }
        else {
            copyIntoBuffer(BODY_NO_DOC_UPSERT_SUFFIX, buffer);
        }
    }
}
@@ -1129,24 +1129,35 @@
 EADER_SUFFIX
+_DOC_UPSERT
  = (%22%5C%22%7D%7D%5Cn%7B
@@ -1161,16 +1161,39 @@
 %5Cn%7B%5C%22doc
+_as_upsert%5C%22:true,%5C%22doc
 %5C%22:%22).ge
@@ -1244,20 +1244,29 @@
  byte%5B%5D 
-BODY
+HEADER_SUFFIX
 _NO_DOC_
@@ -1275,21 +1275,28 @@
 SERT
-_SUFFIX = (%22%7D
+ = (%22%5C%22%7D%7D%5Cn%7B%5C%22doc%5C%22:
 %22).g
@@ -1379,31 +1379,8 @@
 = (%22
-,%5C%22doc_as_upsert%5C%22:true
 %7D%22).
@@ -1764,29 +1764,85 @@
  return 
-HEADER_SUFFIX
+(UPSERT_DOC ? HEADER_SUFFIX_DOC_UPSERT : HEADER_SUFFIX_NO_DOC_UPSERT)
 ;%0A    %7D%0A
@@ -2074,30 +2074,16 @@
 bject) +
- (UPSERT_DOC ?
  BODY_DO
@@ -2108,44 +2108,8 @@
 ngth
- : BODY_NO_DOC_UPSERT_SUFFIX.length)
 ;%0A  
@@ -2290,38 +2290,8 @@
 ody%0A
-        if (UPSERT_DOC) %7B%0A    
     
@@ -2346,106 +2346,8 @@
 r);%0A
-        %7D%0A        else %7B%0A            copyIntoBuffer(BODY_NO_DOC_UPSERT_SUFFIX, buffer);%0A        %7D%0A
     

move "doc_as_update" to the front of the command (eases debugging)