Анализ объединенных, не разделенных XML-сообщений из TCP-потока с использованием С#

Ответ 1

После некоторого времени запирания я думаю, что могу ответить на свой вопрос следующим образом (возможно, я ошибаюсь, исправления приветствуются):

• Я не нашел метода, чтобы XmlReader мог продолжить синтаксический анализ второго XML-сообщения (по крайней мере, нет, если второе сообщение имеет XmlDeclaration). XmlTextReader.ResetState мог бы сделать что-то подобное, но для этого мне пришлось бы принять ту же самую кодировку для всех сообщений. Поэтому я не мог подключить XmlReader непосредственно к TcpStream.

• После закрытия XmlReader буфер не помещается в последнюю позицию читателей. Таким образом, невозможно закрыть читателя и использовать новый, чтобы продолжить следующее сообщение. Я предполагаю, что причина этого заключается в том, что читатель не смог успешно найти все возможные входные потоки.

• Когда XmlReader выдает исключение, невозможно определить, произошло ли это из-за преждевременного EOF или из-за неутвержденного XML. XmlReader.EOF не устанавливается в случае исключения. В качестве обходного решения я получил свой собственный MemoryBuffer, который возвращает последний байт в виде одного байта. Таким образом, я знаю, что XmlReader действительно интересовался последним байтом, и следующее исключение, скорее всего, связано с усеченным сообщением (это своего рода sloppy, поскольку оно может не обнаруживать каждое невербальное сообщение. Однако после добавления большего количества байт в буфер, рано или поздно ошибка будет обнаружена.

• Я мог бы передать мой XmlReader в интерфейс IXmlLineInfo, который дает доступ к LineNumber и LinePosition текущего node. Поэтому после прочтения первого сообщения я помню эти позиции и использую его для усечения буфера. Здесь идет очень неряшливая часть, потому что я должен использовать кодировку символов, чтобы получить позицию байта. Я уверен, что вы можете найти тестовые примеры для кода ниже, где он разбивается (например, внутренние элементы со смешанным кодированием). Но до сих пор он работал для всех моих тестов.

Вот класс парсера, который я придумал, - может ли он быть полезным (я знаю, его очень далеко от совершенства...)

class XmlParser {

    private byte[] buffer = new byte[0];

    public int Length { 
        get {
            return buffer.Length;
        }
    }

    // Append new binary data to the internal data buffer...
    public XmlParser Append(byte[] buffer2) {
        if (buffer2 != null && buffer2.Length > 0) {
            // I know, its not an efficient way to do this.
            // The EofMemoryStream should handle a List<byte[]> ...
            byte[] new_buffer = new byte[buffer.Length + buffer2.Length];
            buffer.CopyTo(new_buffer, 0);
            buffer2.CopyTo(new_buffer, buffer.Length);
            buffer = new_buffer;
        }
        return this;
    }

    // MemoryStream which returns the last byte of the buffer individually,
    // so that we know that the buffering XmlReader really locked at the last
    // byte of the stream.
    // Moreover there is an EOF marker.
    private class EofMemoryStream: Stream {
        public bool EOF { get; private set; }
        private MemoryStream mem_;

        public override bool CanSeek {
            get {
                return false;
            }
        }
        public override bool CanWrite {
            get {
                return false;
            }
        }
        public override bool CanRead {
            get {
                return true;
            }
        }
        public override long Length {
            get { 
                return mem_.Length; 
            }
        }
        public override long Position {
            get {
                return mem_.Position;
            }
            set {
                throw new NotSupportedException();
            }
        }
        public override void Flush() {
            mem_.Flush();
        }
        public override long Seek(long offset, SeekOrigin origin) {
            throw new NotSupportedException();
        }
        public override void SetLength(long value) {
            throw new NotSupportedException();
        }
        public override void Write(byte[] buffer, int offset, int count) {
            throw new NotSupportedException();
        }
        public override int Read(byte[] buffer, int offset, int count) {
            count = Math.Min(count, Math.Max(1, (int)(Length - Position - 1)));
            int nread = mem_.Read(buffer, offset, count);
            if (nread == 0) {
                EOF = true;
            }
            return nread;
        }
        public EofMemoryStream(byte[] buffer) {
            mem_ = new MemoryStream(buffer, false);
            EOF = false;
        }
        protected override void Dispose(bool disposing) {
            mem_.Dispose();
        }

    }

    // Parses the first xml message from the stream.
    // If the first message is not yet complete, it returns null.
    // If the buffer contains non-wellformed xml, it ~should~ throw an exception.
    // After reading an xml message, it pops the data from the byte array.
    public Message deserialize() {
        if (buffer.Length == 0) {
            return null;
        }
        Message message = null;

        Encoding encoding = Message.default_encoding;
        //string xml = encoding.GetString(buffer);

        using (EofMemoryStream sbuffer = new EofMemoryStream (buffer)) {

            XmlDocument xmlDocument = null;
            XmlReaderSettings settings = new XmlReaderSettings();

            int LineNumber = -1;
            int LinePosition = -1;
            bool truncate_buffer = false;

            using (XmlReader xmlReader = XmlReader.Create(sbuffer, settings)) {
                try {
                    // Read to the first node (skipping over some element-types.
                    // Don't use MoveToContent here, because it would skip the
                    // XmlDeclaration too...
                    while (xmlReader.Read() &&
                           (xmlReader.NodeType==XmlNodeType.Whitespace || 
                            xmlReader.NodeType==XmlNodeType.Comment)) {
                    };

                    // Check for XML declaration.
                    // If the message has an XmlDeclaration, extract the encoding.
                    switch (xmlReader.NodeType) {
                        case XmlNodeType.XmlDeclaration: 
                            while (xmlReader.MoveToNextAttribute()) {
                                if (xmlReader.Name == "encoding") {
                                    encoding = Encoding.GetEncoding(xmlReader.Value);
                                }
                            }
                            xmlReader.MoveToContent();
                            xmlReader.Read();
                            break;
                    }

                    // Move to the first element.
                    xmlReader.MoveToContent();

                    if (xmlReader.EOF) {
                        return null;
                    }

                    // Read the entire document.
                    xmlDocument = new XmlDocument();
                    xmlDocument.Load(xmlReader.ReadSubtree());
                } catch (XmlException e) {
                    // The parsing of the xml failed. If the XmlReader did
                    // not yet look at the last byte, it is assumed that the
                    // XML is invalid and the exception is re-thrown.
                    if (sbuffer.EOF) {
                        return null;
                    }
                    throw e;
                }

                {
                    // Try to serialize an internal data structure using XmlSerializer.
                    Type type = null;
                    try {
                        type = Type.GetType("my.namespace." + xmlDocument.DocumentElement.Name);
                    } catch (Exception e) {
                        // No specialized data container for this class found...
                    }
                    if (type == null) {
                        message = new Message();
                    } else {
                        // TODO: reuse the serializer...
                        System.Xml.Serialization.XmlSerializer ser = new System.Xml.Serialization.XmlSerializer(type);
                        message = (Message)ser.Deserialize(new XmlNodeReader(xmlDocument));
                    }
                    message.doc = xmlDocument;
                }

                // At this point, the first XML message was sucessfully parsed.

                // Remember the lineposition of the current end element.
                IXmlLineInfo xmlLineInfo = xmlReader as IXmlLineInfo;
                if (xmlLineInfo != null && xmlLineInfo.HasLineInfo()) {
                    LineNumber = xmlLineInfo.LineNumber;
                    LinePosition = xmlLineInfo.LinePosition;
                }


                // Try to read the rest of the buffer.
                // If an exception is thrown, another xml message appears.
                // This way the xml parser could tell us that the message is finished here.
                // This would be prefered as truncating the buffer using the line info is sloppy.
                try {
                    while (xmlReader.Read()) {
                    }
                } catch {
                    // There comes a second message. Needs workaround for trunkating.
                    truncate_buffer = true;
                }
            }
            if (truncate_buffer) {
                if (LineNumber < 0) {
                    throw new Exception("LineNumber not given. Cannot truncate xml buffer");
                }
                // Convert the buffer to a string using the encoding found before 
                // (or the default encoding).
                string s = encoding.GetString(buffer);

                // Seek to the line.
                int char_index = 0;
                while (--LineNumber > 0) {
                    // Recognize \r , \n , \r\n as newlines...
                    char_index = s.IndexOfAny(new char[] {'\r', '\n'}, char_index);
                    // char_index should not be -1 because LineNumber>0, otherwise an RangeException is 
                    // thrown, which is appropriate.
                    char_index++;
                    if (s[char_index-1]=='\r' && s.Length>char_index && s[char_index]=='\n') {
                        char_index++;
                    }
                }
                char_index += LinePosition - 1;

                var rgx = new System.Text.RegularExpressions.Regex(xmlDocument.DocumentElement.Name + "[ \r\n\t]*\\>");
                System.Text.RegularExpressions.Match match = rgx.Match(s, char_index);
                if (!match.Success || match.Index != char_index) {
                    throw new Exception("could not find EndElement to truncate the xml buffer.");
                }
                char_index += match.Value.Length;

                // Convert the character offset back to the byte offset (for the given encoding).
                int line1_boffset = encoding.GetByteCount(s.Substring(0, char_index));

                // remove the bytes from the buffer.
                buffer = buffer.Skip(line1_boffset).ToArray();
            } else {
                buffer = new byte[0];
            }
        }
        return message;
    }
}

Ответ 2

Чтение в MemoryStream не требуется использовать XmlReader. Вы можете напрямую подключить читателя к потоку, чтобы читать столько, сколько вам нужно, чтобы дойти до конца документа XML. A BufferedStream может использоваться для повышения эффективности чтения из сокета напрямую.

string server = "tcp://myserver"
string message = "GetMyXml"
int port = 13000;
int bufferSize = 1024;

using(var client = new TcpClient(server, port))
using(var clientStream = client.GetStream())
using(var bufferedStream = new BufferedStream(clientStream, bufferSize))
using(var xmlReader = XmlReader.Create(bufferedStream))
{
    xmlReader.MoveToContent();

    try
    {
        while(xmlReader.Read())
        {
            // Check for XML declaration.
            if(xmlReader.NodeType != XmlNodeType.XmlDeclaration)
            {
                throw new Exception("Expected XML declaration.");
            }

            // Move to the first element.
            xmlReader.Read();
            xmlReader.MoveToContent();

            // Read the root element.
            // Hand this document to another method to process further.
            var xmlDocument = XmlDocument.Load(xmlReader.ReadSubtree());
        }
    }
    catch(XmlException ex)
    {
        // Record exception reading stream.
        // Move reader to start of next document or rethrow exception to exit.
    }
}

Ключом к выполнению этой работы является вызов XmlReader.ReadSubtree(), который создает дочерний читатель поверх родительского читателя, который будет обрабатывать текущий элемент (в данном случае корневой элемент) как все дерево XML. Это должно позволить вам отдельно разбирать элементы документа.

Мой код немного неряшлив вокруг чтения документа, тем более, что я игнорирую всю информацию в декларации XML. Я уверен, что есть возможности для улучшения, но, надеюсь, это поможет вам на правильном пути.

Ответ 3

Предполагая, что вы можете изменить протокол, я предлагаю добавлять в сообщения маркеры начала и конца, чтобы при чтении всего текста в виде текстового потока вы могли разделить его на отдельные сообщения (оставляя неполные сообщения в "входящий буфер" какого-либо типа), очистите маркеры, а затем вы знаете, что у вас есть ровно одно сообщение в то время.

Ответ 4

Два вопроса, которые я нашел, были:

• XmlReader будет разрешать только объявление XML в самом начале. Поскольку он не может быть reset, его нужно воссоздать.
• Как только XmlReader выполнил свою работу, он обычно потребляет дополнительные символы после окончания документа, потому что использует метод Read(char[], int, int).

Мое (хрупкое) обходное решение - создать оболочку, которая заполняет массив только до тех пор, пока не встретится " > ". Это удерживает XmlReader от потребляющих символов после окончания > документа, который он разбор:

public class SegmentingReader : TextReader {
    private TextReader reader;
    private char trigger;

    public SegmentingReader(TextReader reader, char trigger) {
        this.reader = reader;
        this.trigger = trigger;
    }

    // Dispose omitted for brevity

    public override int Peek() { return reader.Peek(); }

    public override int Read() { return reader.Read(); }

    public override int Read(char[] buffer, int index, int count) {
        int n = 0;
        while (n < count) {
            char ch = (char)reader.Read();
            buffer[index + n] = ch;
            n++;
            if (ch == trigger) break;
        }
        return n;
    }
}

Затем его можно использовать так же, как:

using(var inputReader = new SegmentingReader(/*TextReader from somewhere */))
using(var serializer = new XmlSerializer(typeof(SerializedClass)))
while (inputReader.Peek() != -1)
{
    using (var xmlReader = XmlReader.Create(inputReader)) {
        xmlReader.MoveToContent();
        var obj = serializer.Deserialize(xmlReader.ReadSubtree());
        DoStuff(obj);
    }
}