<commit_before>/**
 * Copyright (C) 2014 Stratio (http://stratio.com)
 *
 * 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.stratio.sparkta.plugin.parser.datetime

import java.io.{Serializable => JSerializable}
import java.util.Date

import com.stratio.sparkta.sdk.ValidatingPropertyMap._
import com.stratio.sparkta.sdk.{Event, Parser}
import org.joda.time.format.{DateTimeFormatter, ISODateTimeFormat}

class DateTimeParser(name: String,
                     order: Integer,
                     inputField: String,
                     outputFields: Seq[String],
                     properties: Map[String, JSerializable])
  extends Parser(name, order, inputField, outputFields, properties) {

  private final val InputFormatterLabel = "inputFormat"

  private val formatter = {
    val formats = properties.toSeq.map(x => (x._1, x._2.toString)).toMap
    val formatMethods = classOf[ISODateTimeFormat].getMethods.toSeq.map(x => (x.getName, x)).toMap
    formats.get(InputFormatterLabel) match {
      case Some(format) => {
        format match {
          case "unix" => Some(Right("unix"))
          case "unixMillis" => Some(Right("unixMillis"))
          case "autoGenerated" => Some(Right("autoGenerated"))
          case _ => Some(Left(formatMethods(format).invoke(None).asInstanceOf[DateTimeFormatter]))
        }
      }
      case None => None
    }
  }

  override def parse(data: Event): Event = {
    new Event(data.keyMap.flatMap {
      case (key, value) =>
        outputFields.map(outputField => {
          if (inputField == key && formatter.isDefined && !value.isInstanceOf[Date]) {
            formatter.get match {
              case Right("unix") =>
                (outputField, new Date(value.toString.toLong * 1000L))
              case Right("unixMillis") =>
                (outputField, new Date(value.toString.toLong))
              case Right("autoGenerated") =>
                (outputField, new Date())
              case Left(formatter) =>
                (outputField, formatter.parseDateTime(value.toString).toDate)
            }
          } else {
            (key, value)
          }
        })
    })
  }

}


<commit_msg>DateTime parser serialization exception fixed<commit_after>/**
 * Copyright (C) 2015 Stratio (http://stratio.com)
 *
 * 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.stratio.sparkta.plugin.parser.datetime

import java.io.{Serializable => JSerializable}
import java.util.Date

import org.joda.time.format.{DateTimeFormatter, ISODateTimeFormat}

import com.stratio.sparkta.sdk.{Event, Parser}

class DateTimeParser(name: String,
                     order: Integer,
                     inputField: String,
                     outputFields: Seq[String],
                     properties: Map[String, JSerializable])
  extends Parser(name, order, inputField, outputFields, properties) {

  val formats = properties.get("inputFormat")

  override def parse(data: Event): Event = {
    new Event(data.keyMap.flatMap {
      case (key, value) => outputFields.map(outputField => {
        val formatterOption = DateTimeParser.formatter(formats)
        if (inputField == key && formatterOption.isDefined && !value.isInstanceOf[Date]) {
          formatterOption.get match {
            case Right("unix") =>
              (outputField, new Date(value.toString.toLong * 1000L))
            case Right("unixMillis") =>
              (outputField, new Date(value.toString.toLong))
            case Right("autoGenerated") =>
              (outputField, new Date())
            case Left(formatter) =>
              (outputField, formatter.parseDateTime(value.toString).toDate)
          }
        } else {
          (key, value)
        }
      })
    })
  }
}

object DateTimeParser {

  val formatMethods = classOf[ISODateTimeFormat].getMethods.toSeq.map(x => (x.getName, x)).toMap

  def formatter(formats: Option[JSerializable]): Option[Either[DateTimeFormatter, String]] = {
    formats match {
      case Some(format) => {
        format.toString match {
          case "unix" => Some(Right("unix"))
          case "unixMillis" => Some(Right("unixMillis"))
          case "autoGenerated" => Some(Right("autoGenerated"))
          case _ => Some(Left(formatMethods(format.toString).invoke(None).asInstanceOf[DateTimeFormatter]))
        }
      }
      case None => None
    }
  }
}

{'fraction_non_alphanumeric': 0.112, 'fraction_numerical': 0.0048372093023255815, 'mean_word_length': 2.689773507206589, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4906857', 'n_tokens_mistral': 1497, 'n_tokens_neox': 1456, 'n_words': 394}