1 files changed, 105 insertions, 8 deletions

diff --git a/src/main/java/com/lumaserv/bgp/BGPSession.java b/src/main/java/com/lumaserv/bgp/BGPSession.java
index 3ffadd8..94a4f5a 100644
--- a/src/main/java/com/lumaserv/bgp/BGPSession.java
+++ b/src/main/java/com/lumaserv/bgp/BGPSession.java
@@ -1,12 +1,15 @@
 package com.lumaserv.bgp;
 
 import com.lumaserv.bgp.protocol.BGPPacket;
+import com.lumaserv.bgp.protocol.message.BGPNotification;
+import com.lumaserv.bgp.protocol.message.BGPOpen;
 import com.lumaserv.bgp.protocol.message.BGPUpdate;
 import lombok.Getter;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketException;
 
@@ -14,15 +17,32 @@ public class BGPSession implements Runnable {
 
     @Getter
     final BGPSessionConfiguration configuration;
-    final InputStream inputStream;
-    final OutputStream outputStream;
+    final String host;
+    final int port;
+    Socket socket;
+    InputStream inputStream;
+    OutputStream outputStream;
     @Getter
     boolean closed;
+    boolean outgoing;
+    BGPFsm fsm;
+    Thread thread;
 
-    public BGPSession(Socket socket, BGPSessionConfiguration configuration) throws IOException {
+    public BGPSession(Socket socket, BGPSessionConfiguration configuration, BGPFsm fsm) throws IOException {
         this.configuration = configuration;
+        this.socket = socket;
         this.inputStream = socket.getInputStream();
         this.outputStream = socket.getOutputStream();
+        this.fsm = fsm;
+        this.host = null;
+        this.port = -1;
+    }
+
+    public BGPSession(BGPSessionConfiguration configuration, BGPFsm fsm, String host, int port) throws IOException {
+        this.configuration = configuration;
+        this.fsm = fsm;
+        this.host = host;
+        this.port = port;
     }
 
     public void keepAlive() {
@@ -36,28 +56,105 @@ public class BGPSession implements Runnable {
     private void handle(BGPPacket packet) {
         switch (packet.getType()) {
             case NOTIFICATION:
-                closed = true;
-                configuration.getListener().onClose(this);
+                BGPNotification notif = new BGPNotification(packet.getMessage());
+                if (notif.getMajorErrorCode() == BGPNotification.Error.MESSAGE_HEADER_ERROR.getCode() &&
+                    notif.getMinorErrorCode() == BGPNotification.OpenMessageError.UNSUPPORTED_VERSION_NUMBER.getCode())
+                    fsm.getCurrentState().notifMsgVerErr(notif);
+                else
+                    fsm.getCurrentState().notifMsg(notif);
                 break;
             case KEEPALIVE:
-                keepAlive();
+                fsm.getCurrentState().keepAliveMsg();
+                //keepAlive();
+                break;
+            case OPEN:
+                BGPOpen open = new BGPOpen(packet.getMessage());
+                fsm.getCurrentState().openMsg(open);
                 break;
             case UPDATE: {
-                configuration.getListener().onUpdate(this, new BGPUpdate(packet.getMessage()));
+                BGPUpdate update = new BGPUpdate(packet.getMessage());
+                fsm.getCurrentState().updateMsg(update);
+                configuration.getListener().onUpdate(this, update);
                 break;
             }
         }
     }
 
+    public void startIncoming() {
+        outgoing = false;
+        thread = new Thread(this);
+        thread.start();
+    }
+
+    public void startOutgoing() {
+        outgoing = true;
+        thread = new Thread(this);
+        thread.start();
+    }
+
+    public void sendOpen(BGPOpen request) throws IOException {
+        outputStream.write(new BGPPacket().setType(BGPPacket.Type.OPEN).setMessage(request.build()).build());
+System.out.println("Sent open");
+    }
+
+    public void retryConnection() {
+        if (!socket.isConnected()) {
+            try {
+                socket.close();
+            } catch (IOException ex) {
+                ex.printStackTrace();
+                throw new RuntimeException(ex);
+            }
+        }
+    }
+
+    public void dropConnection() {
+        closed = true;
+        try {
+            if (outputStream != null)
+                outputStream.close();
+            if (inputStream != null)
+                inputStream.close();
+            socket.close();
+        } catch (IOException ex) {
+            ex.printStackTrace();
+            throw new RuntimeException(ex);
+        }
+    }
+
     public void run() {
         try {
+            if (socket == null) {
+                while (true) {
+                    try {
+                        System.out.println("Outgoing");
+                        socket = new Socket();
+                        socket.connect(new InetSocketAddress(host, port));
+                        break;
+                    } catch(IOException ex) {
+                        if (closed) {
+                            throw new IOException("Closed", ex);
+                        }
+                    }
+                }
+                System.out.println("Outgoing accepted");
+                inputStream = socket.getInputStream();
+                outputStream = socket.getOutputStream();
+                fsm.getCurrentState().tcpCRAcked();
+            } else {
+                System.out.println("Incoming");
+                fsm.getCurrentState().tcpConnectionConfirmed();
+            }
             while (!closed) {
                 handle(BGPPacket.read(inputStream));
             }
+            System.out.println("Closed!");
         } catch (IOException ex) {
+            if (!closed)
+                fsm.getCurrentState().tcpConnectionFails();
+            ex.printStackTrace();
             closed = true;
             configuration.getListener().onClose(this);
         }
     }
-
 }