<commit_before>package com.socrata.datacoordinator.backup

import com.socrata.datacoordinator.packets.{Packet, PacketOutputStream}
import java.nio.ByteBuffer
import com.socrata.datacoordinator.id.{RowId, DatasetId}
import com.socrata.datacoordinator.truth.loader.Delogger
import java.io._
import com.socrata.datacoordinator.common.util.ByteBufferInputStream
import com.rojoma.json.util.JsonUtil
import com.socrata.datacoordinator.truth.metadata.{ColumnInfo, CopyInfo}
import scala.Some

class Protocol[LogData](logDataCodec: Codec[LogData]) {
  import Packet.{SimplePacket, LabelledPacket}

  // primary -> backup
  object NothingYet extends SimplePacket("nothing yet")
  object DataDone extends SimplePacket("data done")
  object WillResync extends SimplePacket("will resync")

  object DatasetUpdated extends LabelledPacket("dataset updated") {
    private val label = "dataset updated"

    def apply(id: DatasetId, version: Long) =
      create { os =>
        val dos = new DataOutputStream(os)
        dos.writeLong(id.underlying)
        dos.writeLong(version)
        dos.flush()
      }

    def unapply(packet: Packet): Option[(DatasetId, Long)] =
      extract(packet) map { data =>
        if(data.remaining != 16) throw new PacketDecodeException("DatasetUpdate packet does not contain 16 bytes of payload")
        val id = new DatasetId(data.getLong())
        val version = data.getLong()
        (id, version)
      }
  }

  object LogData extends LabelledPacket("log data") {
    def apply(event: LogData) = create { os =>
      val dos = new DataOutputStream(os)
      dos.write("log data|".getBytes)
      logDataCodec.encode(dos, event)
      dos.flush()
    }

    def unapply(packet: Packet): Option[LogData] = extract(packet) map { data =>
      val stream = new DataInputStream(new ByteBufferInputStream(data))
      logDataCodec.decode(stream)
    }
  }

  // backup -> primary
  object OkStillWaiting extends SimplePacket("ok still waiting")
  object AlreadyHaveThat extends SimplePacket("already have that")
  object ResyncRequired extends SimplePacket("resync required")
  object WillingToAccept extends SimplePacket("willing to accept")
  object AcknowledgeReceipt extends SimplePacket("acknowledged")
}

class PacketDecodeException(msg: String) extends Exception(msg)
<commit_msg>Remove unused label, add stream labels for protocol<commit_after>package com.socrata.datacoordinator.backup

import com.socrata.datacoordinator.packets.{Packet, PacketOutputStream}
import java.nio.ByteBuffer
import com.socrata.datacoordinator.id.{RowId, DatasetId}
import com.socrata.datacoordinator.truth.loader.Delogger
import java.io._
import com.socrata.datacoordinator.common.util.ByteBufferInputStream
import com.rojoma.json.util.JsonUtil
import com.socrata.datacoordinator.truth.metadata.{ColumnInfo, CopyInfo}
import scala.Some

class Protocol[LogData](logDataCodec: Codec[LogData]) {
  import Packet.{SimplePacket, LabelledPacket}

  // primary -> backup
  object NothingYet extends SimplePacket("nothing yet")
  object DataDone extends SimplePacket("data done")
  object WillResync extends SimplePacket("will resync")

  object DatasetUpdated extends LabelledPacket("dataset updated") {
    def apply(id: DatasetId, version: Long) =
      create { os =>
        val dos = new DataOutputStream(os)
        dos.writeLong(id.underlying)
        dos.writeLong(version)
        dos.flush()
      }

    def unapply(packet: Packet): Option[(DatasetId, Long)] =
      extract(packet) map { data =>
        if(data.remaining != 16) throw new PacketDecodeException("DatasetUpdate packet does not contain 16 bytes of payload")
        val id = new DatasetId(data.getLong())
        val version = data.getLong()
        (id, version)
      }
  }

  object LogData extends LabelledPacket("log data") {
    def apply(event: LogData) = create { os =>
      val dos = new DataOutputStream(os)
      dos.write("log data|".getBytes)
      logDataCodec.encode(dos, event)
      dos.flush()
    }

    def unapply(packet: Packet): Option[LogData] = extract(packet) map { data =>
      val stream = new DataInputStream(new ByteBufferInputStream(data))
      logDataCodec.decode(stream)
    }
  }

  // backup -> primary
  object OkStillWaiting extends SimplePacket("ok still waiting")
  object AlreadyHaveThat extends SimplePacket("already have that")
  object ResyncRequired extends SimplePacket("resync required")
  object WillingToAccept extends SimplePacket("willing to accept")
  object AcknowledgeReceipt extends SimplePacket("acknowledged")

  val StreamDataLabel = "d"
  val StreamEndLabel = "end of stream"
}

class PacketDecodeException(msg: String) extends Exception(msg)

{'fraction_non_alphanumeric': 0.09780055519965834, 'fraction_numerical': 0.001708306641042067, 'mean_word_length': 4.192904656319291, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4156519', 'n_tokens_mistral': 1381, 'n_tokens_neox': 1334, 'n_words': 342}