class Kv78Parser {

    public static Kv78Packet parseMessage(String input) {
        Kv78Packet p = new Kv78Packet();
        Kv78Table t = null;
        String[] lines = input.split(CHAR_NEWLINE);
        String[] headers = null;
        for (String line : lines) {
            if (line.startsWith(CHAR_CONTROL)) {
                String[] columns = line.substring(2).split("\\|");
                switch (line.substring(1,2)) {
                    case "G":
                        p.setType(columns[0]);
                        p.setComment(columns[2]);
                        p.setEncoding(columns[5]);
                        p.setVersion(columns[6]);
                        p.setGenerated(ZonedDateTime.parse(columns[7]));
                        break;
                    case "T":
                        if (t != null) {
                            p.addTable(t);
                        }
                        t = new Kv78Table();
                        t.setTableName(columns[0]);
                        t.setTableComment(columns[2]);
                        break;
                    case "L":
                        headers = columns;
                        break;
                    default:
                        break;
                }
            } else {
                if (headers != null) {
                    Map<String, String> record = new HashMap<>();
                    String[] cells = line.split(CHAR_SPLIT);
                    for (int i = 0; i < cells.length; i++) {
                        if (!CHAR_EMPTY.contentEquals(cells[i])) {
                            record.put(headers[i], cells[i]);
                        }
                    }
                    if (t == null) {
                        t = new Kv78Table();
                    }
                    t.getRecords().add(record);
                } else {
                    // Invalid message!
                    LOGGER.error("Failed to parse message");
                }
            }
        }
        p.addTable(t);
        return p;
    }

}

class Kv78ParserTest {

    @Test
    public void parseGeneralMessage() {

        String data = "\\GKV8turbo_generalmessages|KV8turbo_generalmessages|openOV RET|||UTF-8|0.1|2017-04-11T21:53:25+02:00|\uFEFF\r\n" +
                "\\TGENERALMESSAGEUPDATE|GENERALMESSAGEUPDATE|start object\r\n" +
                "\\LDataOwnerCode|MessageCodeDate|MessageCodeNumber|TimingPointDataOwnerCode|TimingPointCode|MessageType|MessageDurationType|MessageStartTime|MessageEndTime|MessageContent|ReasonType|SubReasonType|ReasonContent|EffectType|SubEffectType|EffectContent|MeasureType|SubMeasureType|MeasureContent|AdviceType|SubAdviceType|AdviceContent|MessageTimeStamp\r\n" +
                "RET|2017-04-11|27|ALGEMEEN|31001347|GENERAL|REMOVE|2017-04-11T21:51:08+02:00|\\0|Door defect materieel is er op tram 23 een rit uitgevallen. Houd rekening met een extra reistijd tot 15 min.|0|0|\\0|0|0|\\0|0|6|\\0|0|0|\\0|2017-04-11T21:53:24+02:00\r\n";
        Kv78Packet p = Kv78Parser.parseMessage(data);

        Assert.assertEquals("KV8turbo_generalmessages", p.getType());
        Assert.assertEquals("openOV RET", p.getComment());
        Assert.assertEquals("UTF-8", p.getEncoding());
        Assert.assertEquals("0.1", p.getVersion());
        Assert.assertEquals(ZonedDateTime.parse("2017-04-11T21:53:25+02:00"), p.getGenerated());

        Assert.assertEquals("GENERALMESSAGEUPDATE", p.getTables().get(0).getTableName());
        Assert.assertEquals("start object", p.getTables().get(0).getTableComment());

        Assert.assertEquals(1, p.getTables().get(0).getRecords().size());
        Map<String, String> record = p.getTables().get(0).getRecords().get(0);
        Assert.assertEquals(18, record.size());
        Assert.assertEquals("RET", record.get("DataOwnerCode"));
        Assert.assertEquals("2017-04-11", record.get("MessageCodeDate"));
        Assert.assertEquals("27", record.get("MessageCodeNumber"));
        Assert.assertEquals("ALGEMEEN", record.get("TimingPointDataOwnerCode"));
        Assert.assertEquals("31001347", record.get("TimingPointCode"));
        Assert.assertEquals("GENERAL", record.get("MessageType"));
        Assert.assertEquals("REMOVE", record.get("MessageDurationType"));
        Assert.assertEquals("2017-04-11T21:51:08+02:00", record.get("MessageStartTime"));
        Assert.assertEquals("Door defect materieel is er op tram 23 een rit uitgevallen. Houd rekening met een extra reistijd tot 15 min.", record.get("MessageContent"));
        Assert.assertEquals("0", record.get("ReasonType"));
        Assert.assertEquals("0", record.get("SubReasonType"));
        Assert.assertEquals("0", record.get("EffectType"));
        Assert.assertEquals("0", record.get("SubEffectType"));
        Assert.assertEquals("0", record.get("MeasureType"));
        Assert.assertEquals("6", record.get("SubMeasureType"));
        Assert.assertEquals("0", record.get("AdviceType"));
        Assert.assertEquals("0", record.get("SubAdviceType"));
        Assert.assertEquals("2017-04-11T21:53:24+02:00", record.get("MessageTimeStamp"));


    }

}