Re-implement stream open and close

This is more suitable for the new RX concept.

1 files changed, 18 insertions, 3 deletions

diff --git a/xmpp/session.go b/xmpp/session.go
index f18fd2a..4be3386 100644
--- a/xmpp/session.go
+++ b/xmpp/session.go
@@ -39,8 +39,6 @@ func StartSession(out chan<- any, jid string, pwd string) (in chan<- any) {
 }
 
 func (s *session) run() {
-	defer func() { s.out <- SessionDisconnect{} }()
-
 	err := s.startTransport()
 	if err != nil {
 		return
@@ -55,8 +53,25 @@ func (s *session) run() {
 	lw := logger{"[TX] "}
 	w := io.MultiWriter(s.transport, lw)
 	s.tx = xml.NewEncoder(w)
+	defer s.tx.Close()
+
+	openStream(s.tx, s.jid)
+	defer closeStream(s.tx)
 
-	s.out <- SessionConnect{}
+	s.out <- SessionConnect{}  // TODO this should be sent after initial presence
+	defer func() { s.out <- SessionDisconnect{} }()
+
+	for {
+		select {
+		case e := <-s.rx:
+			log.Print(e)
+		case signal := <-s.in:
+			switch signal.(type) {
+			case SessionShouldDisconnect:
+				return
+			}
+		}
+	}
 }
 
 func (s *session) startTransport() error {