From 4d9633f05b70a15aeb0ff39a405d76bcc2d66464 Mon Sep 17 00:00:00 2001 From: xengineering Date: Wed, 28 Jun 2023 09:57:55 +0200 Subject: xmpp: Move Stream logic to stream_pair.go The new source file should contain the complete stream logic. --- xmpp/stream_pair.go | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 52 insertions(+) create mode 100644 xmpp/stream_pair.go (limited to 'xmpp/stream_pair.go') diff --git a/xmpp/stream_pair.go b/xmpp/stream_pair.go new file mode 100644 index 0000000..6224a77 --- /dev/null +++ b/xmpp/stream_pair.go @@ -0,0 +1,52 @@ +package xmpp + +import ( + "encoding/xml" + "log" +) + +func runStreamPair(s *session) { + openStream(s) + defer closeStream(s) + + for { + select { + case data := <-s.in: + switch data.(type) { + case SessionShouldDisconnect: + return + default: + log.Printf("Unknown data '%d'!\n", data) + } + case _ = <-s.rx: + // TODO route received XML token here + } + } +} + +func openStream(s *session) { + start := xml.StartElement{ + xml.Name{"jabber:client", "stream:stream"}, + []xml.Attr{ + xml.Attr{xml.Name{"", "from"}, s.jid}, + xml.Attr{xml.Name{"", "to"}, domainpart(s.jid)}, + xml.Attr{xml.Name{"", "version"}, "1.0"}, + xml.Attr{xml.Name{"", "xml:lang"}, "en"}, + xml.Attr{xml.Name{"", "xmlns:stream"}, "http://etherx.jabber.org/streams"}, + }, + } + + s.streamEnd = start.End() + + err := s.ed.encodeToken(start) + if err != nil { + log.Println("Could not encode stream start!") + } +} + +func closeStream(s *session) { + err := s.ed.encodeToken(s.streamEnd) + if err != nil { + log.Println("Could not encode stream end!") + } +} -- cgit v1.2.3-70-g09d2 From fdddefe08fa8c2a7aeab2b5a865fcb3791ab6080 Mon Sep 17 00:00:00 2001 From: xengineering Date: Wed, 28 Jun 2023 10:22:18 +0200 Subject: xmpp: Implement syncStreams() This should ensure that the incoming and outgoing XML streams are in sync. --- xmpp/stream_pair.go | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) (limited to 'xmpp/stream_pair.go') diff --git a/xmpp/stream_pair.go b/xmpp/stream_pair.go index 6224a77..e02cca6 100644 --- a/xmpp/stream_pair.go +++ b/xmpp/stream_pair.go @@ -42,6 +42,36 @@ func openStream(s *session) { if err != nil { log.Println("Could not encode stream start!") } + + syncStreams(s) +} + +// syncStreams drops XML tokens from the receiving stream until an +// xml.StartElement with the local name `stream` is received. If this function +// is called after opening a new stream in the sending direction it is ensured +// that both streams directions work on the same stream level and are in sync. +// Tokens received which are not a stream StartElement are not handled but +// logged since this should not happen. +func syncStreams(s *session) { + for { + select { + case data := <-s.in: + switch data.(type) { + case SessionShouldDisconnect: + return + default: + log.Printf("Unhandled data '%d' during stream sync!\n", data) + } + case t := <-s.rx: + switch token := t.(type) { + case xml.StartElement: + if token.Name.Local == "stream" { + return + } + } + log.Printf("Unhandled XML token '%v' during stream sync!\n", t) + } + } } func closeStream(s *session) { -- cgit v1.2.3-70-g09d2