osmo: Improve the connection class for re-connect and process handling

Use the usual rx/tx process handling, re-connect, start/stop

Makefile

@@ -5,6 +5,7 @@ GST_CONVERT = gst-convert
 CONVERT_RULES = -r'Osmo.LogManager->LogManager' \
 		-r'Osmo.LogArea->LogArea'  \
 		-r'Osmo.LogLevel->LogLevel' \
+                -r'Osmo.TimerScheduler->TimerScheduler' \
                 -r'Sockets.StreamSocket->SocketStream' \
 		-r'DateTime->DateAndTime' \
 		-r'(Duration milliseconds: ``@args1) -> (Duration milliSeconds: ``@args1)' \

osmo/OsmoAppConnection.st

@@ -17,12 +17,70 @@
 "
 
 Object subclass: OsmoAppConnection [
-    | socket writeQueue demuxer muxer dispatcher token |
+    | socket writeQueue demuxer muxer dispatcher token hostname port tx_proc rx_proc started |
     <category: 'OsmoNetwork-Socket'>
     <comment: 'I connect to a OpenBSC App on the Control Port and wait for
 TRAPS coming from the server and will act on these.
 TODO: re-use the IPADispatcher across connections.'>
 
+    OsmoAppConnection class >> connectionException [
+        <category: 'exceptions'>
+        ^ SystemExceptions.FileError
+    ]
+
+    OsmoAppConnection class >> new [
+        ^(self basicNew)
+            hostname: '127.0.0.1';
+            port: 4250;
+            yourself
+    ]
+
+    driveDispatch [
+        <category: 'private'>
+
+        [
+            self dispatchOne
+        ] on: SystemExceptions.EndOfStream do: [:e |
+            self logError: 'OsmoApplication eof' area: #osmo.
+            self scheduleReconnect
+        ] on: SystemExceptions.FileError do: [:e |
+            self logError: 'OsmoApplication file-error' area: #osmo.
+            self scheduleReconnect
+        ] on: Error do:  [:e |
+            e logException: 'OsmoApplication error' area: #osmo.
+            self scheduleReconnect
+        ]
+    ]
+
+    driveSend [
+        <category: 'private'>
+        [
+            self sendOne
+        ] on: SystemExceptions.EndOfStream do: [:e |
+            self logError: 'OsmoApplication eof' area: #osmo.
+            self scheduleReconnect
+        ] on: Error do:  [:e |
+            e logException: 'OsmoApplication error' area: #osmo.
+            self scheduleReconnect
+        ]
+    ]
+
+    reconnect [
+        <category: 'private'>
+        self logNotice: 'Going to reconnect socket' area: #osmo.
+        self terminate.
+        started ifTrue: [self start]
+    ]
+
+    scheduleReconnect [
+        <category: 'private'>
+        socket ifNotNil: [socket close. socket := nil].
+        TimerScheduler instance scheduleInSeconds: 1 block: [self reconnect].
+        "We are done now"
+        Processor activeProcess terminate
+    ]
+
+
     initializeDispatcher [
         "Allow another class to register handlers"
     ]
@@ -31,44 +89,72 @@ TODO: re-use the IPADispatcher across connections.'>
         token := aToken.
     ]
 
+    hostname: aHostname [
+        <category: 'creation'>
+        hostname := aHostname
+    ]
+
+    port: aPort [
+        <category: 'creation'>
+        port := aPort
+    ]
+
     createConnection: aHostname port: aPort [
         <category: 'socket'>
         ^ Sockets.Socket remote: aHostname port: aPort.
     ]
 
-    connect: aPort [
+    connect [
         <category: 'connect'>
-        ^ self connect: '127.0.0.1' port: aPort
-    ]
-
-    connect: aHostname port: aPort [
         | ipa |
-        <category: 'connect'>
-
         socket ifNotNil: [socket close].
-
-        socket := self createConnection: aHostname port: aPort.
+        socket := self createConnection: hostname port: port.
         writeQueue := SharedQueue new.
-
         demuxer := IPADemuxer initOn: socket.
         muxer := IPAMuxer initOn: writeQueue.
-
         dispatcher := IPADispatcher new.
         dispatcher initialize.
         self initializeDispatcher.
-
-        ipa := IPAProtoHandler new
-                registerOn: dispatcher;
-                muxer: muxer;
-                token: token;
-                yourself.
+        ipa := (IPAProtoHandler new)
+                    registerOn: dispatcher;
+                    muxer: muxer;
+                    token: token;
+                    yourself
     ]
 
-    connect [
-	<category: 'connect'>
-	^ self connect: 4250.
+    start [
+        <category: 'connect'>
+        started := true.
+
+        [
+            self logNotice: 'Attempting to connect' area: #osmo.
+            self connect
+        ] on: self class connectionException do: [
+            self logError: 'Failed to connect.' area: #osmo.
+            ^Osmo.TimerScheduler instance scheduleInSeconds: 1 block: [self reconnect]].
+
+        rx_proc :=
+            [Processor activeProcess name: 'OsmoAppConnection-RX'.
+            [self driveDispatch] repeat] fork.
+        tx_proc := [Processor activeProcess name: 'OsmoAppConnection-TX'.
+            [self driveSend] repeat] fork
     ]
 
+    stop [
+        <category: 'connect'>
+        started := false.
+        self terminate
+        "A reconnect timer might be running right now"
+    ]
+
+    terminate [
+        <category: 'connect'>
+        tx_proc ifNotNil: [tx_proc terminate].
+        rx_proc ifNotNil: [rx_proc terminate].
+        socket ifNotNil: [socket close. socket := nil]
+    ]
+
+
     sendOne [
         | msg |
         <category: 'dispatch'>

pharo-porting/changes_for_pharo.st

@@ -19,7 +19,38 @@ OsmoUDPSocket extend [
 
 OsmoAppConnection extend [
     createConnection: aHostname port: aPort [
-        <category: 'pharo-porting'>
-        ^ SocketStream openConnectionToHostNamed: aHostname port: aPort
+	<category: 'pharo-porting'>
+	^(SocketStream openConnectionToHostNamed: aHostname port: aPort)
+	    noTimeout;
+	    yourself
+    ]
+
+    driveDispatch [
+        <category: 'private'>
+        [
+            [self dispatchOne] on: ConnectionClosed do: [:e |
+                self logError: 'OsmoApplication dispatch failed.' area: #osmo.
+                self scheduleReconnect]
+        ] on: Error do: [:e |
+                e logException: 'OsmoApplication error' area: #osmo.
+                self scheduleReconnect]
+    ]
+
+    driveSend [
+        <category: 'private'>
+
+        [
+            self sendOne.
+        ] on: ConnectionClosed do: [:e |
+            e logException: 'OsmoAppConnection send failed' area: #osmo.
+            self scheduleReconnect.
+        ]
+    ]
+]
+
+OsmoAppConnection class extend [
+    connectionException [
+	<category: 'pharo-porting'>
+	^ConnectionTimedOut
     ]
 ]