Untitled

// Contains the implementation of a LSP server.

package lsp

import (
	"errors"
	"github.com/cmu440/lspnet"
	"fmt" // need to delete this
	"container/list"
	"encoding/json" // for json Marshal
)


type pair struct {
    pair_id, pair_payload interface{} 
}


type server struct {
	connection *lspnet.UDPConn // ?

	// [EpochLimit: 5, EpochMillis: 2000, WindowSize: 1, MaxBackOffInterval: 0,MaxUnackedMessages: 1]
	EpochLimit int
	EpochMillis int
	WindowSize int
	MaxBackOffInterval int
	MaxUnackedMessages int

	client_cur_index int // current client_id number (e.g. current there are 5 clients already, then the number here can be 6, which means the next client will have an id value 6)
	m map[int]client_2 // map conn_id to client struct!! // keep track of client's IP address+port with its client_id (so if resend connection message, we do not generate a new client_id??)
	
	queue *list.List // use list so the size will not be fixed
}


type client_2 struct {
	client_id int
	server_sequence_number int // server's copy of sequence number for a specific client
	client_sequence_number int // server's expecting client's sequence number?
	ack_status int // which message it has acked?

    // client's addr
    client_addr *lspnet.UDPAddr
    // map (unordered data)
    unordered_queue map[int]string // type?

	// store un-ordered data message (priority queue, may use a heap) (for seq num that are too large, so we can not yet put into the queue)
	// priority_queue heap.Heap ??
	// probably also the sequence of message to send back to the client???
}


// NewServer creates, initiates, and returns a new server. This function should
// NOT block. Instead, it should spawn one or more goroutines (to handle things
// like accepting incoming client connections, triggering epoch events at
// fixed intervals, synchronizing events using a for-select loop like you saw in
// project 0, etc.) and immediately return. It should return a non-nil error if
// there was an error resolving or listening on the specified port number.
func NewServer(port int, params *Params) (Server, error) {
	// create a new server and initialize
	var LSPserver server

	// fmt.Println("port = ", port)
	// fmt.Println("params = ", params)
	port_string := fmt.Sprintf("%d", port)
    fmt.Println("port_string = ", port_string)


	addr_string := "127.0.0.1: " + port_string

    // addr_string := JoinHostPort("127.0.0.1", port_string)

	a, error := lspnet.ResolveUDPAddr("udp", addr_string)
	if error != nil {
        fmt.Println(error)
        return &LSPserver, error
    }

	connection, error := lspnet.ListenUDP("udp", a)
	if error != nil {
        fmt.Println(error)
        return &LSPserver, error
    }

    LSPserver.connection = connection
    LSPserver.EpochLimit = params.EpochLimit
    LSPserver.EpochMillis = params.EpochMillis
    LSPserver.WindowSize = params.WindowSize
    LSPserver.MaxBackOffInterval = params.MaxBackOffInterval
    LSPserver.MaxUnackedMessages = params.MaxUnackedMessages

    LSPserver.m = make(map[int]client_2) // !!!

    LSPserver.queue = list.New() // ??????
    

    fmt.Println("Connection sets")

	// start main goroutine
	// go LSPserver.main_routine()

	// start read goroutine
	go LSPserver.read_routine() // ??????? always reading

	return &LSPserver, nil
}




/*

This is where we turn UDP into LSP

We continue to read message, do some processing, and store it in the queue.
For the 3 different cases, we need to do different processing and send it to the main routine

we need to know the latest sequence number that is put into the queue, and only the number that is continuously larger than the current sequence number can be added to the queue

we do not need to call read_routine again and again, because it runs when server starts and never ends
Just continue to send sth into the queue

We just need tone read_routine
*/
func (s *server) read_routine(){
	fmt.Println("in read_routine")
    s.client_cur_index += 1

	for {
        fmt.Println("in read_routine!!!!!!")
        message_temp := make([]byte, 1000)
        // message_read, _, _ := s.connection.Read(message_temp[:]) // ???
        n, addr_, _ := s.connection.ReadFromUDP(message_temp) // ???
        fmt.Println("message_temp = ", string(message_temp))
        // message_read, _, err := s.connection.readFromUDP(message_temp[:]) // ???

        // convert from byte[] bacl to Message!
        var message Message // ??????

        error := json.Unmarshal(message_temp[:n], &message) // need to read limited length
        if error != nil{
        	fmt.Println("error in Unmarshal")
        	return 
        }

        // fmt.Println("message_read = ", message_read)
        fmt.Println("message after unmarshal = ", message)

        if message.Type == 0{ // or 0? connect case (do not need to forward to main routine?)  lspnet.MsgConnect?
        	fmt.Println("Type Connect")

        	var new_client client_2 
        	new_client.client_id = s.client_cur_index
        	new_client.server_sequence_number = message.SeqNum
        	new_client.client_sequence_number = message.SeqNum

            new_client.client_addr = addr_ // added

        	// also need to initialize a priority queue? (no for now, and we can just use a sliding window to achieve this)

        	s.m[s.client_cur_index] = new_client
        	s.client_cur_index += 1

        	// send ack message to client (with client's connect_id)
        	ack_message := NewAck(new_client.client_id, message.SeqNum)
        	message_bytes, error := json.Marshal(ack_message)
    		if error != nil{
    			return 
    		}

            fmt.Println("server new connection before write to udp")

            s.connection.WriteToUDP(message_bytes, addr_) // 需要看會不會block (如果會block就用write_routine)

            fmt.Println("server new connection after write to udp")
        	
        }

        if message.Type == 1{ // data (按照順序傳進list/queue)
            fmt.Println("Type Data")

        	this_seq_num := message.SeqNum
        	this_client_id := message.ConnID

            // check checksum and size first
            this_size := message.Size
            this_payload := message.Payload
            fmt.Println("message.Size = ", message.Size)
            fmt.Println("len of payload = ", len(message.Payload))

            if len(this_payload) != this_size{
                fmt.Println("the size is incorrect")
                return 
            }

            // checksum
            this_checksum := CalculateChecksum(message.ConnID, message.SeqNum, message.Size, message.Payload)
            fmt.Println("calculated checksum = ", this_checksum)
            fmt.Println("message.Checksum = ", message.Checksum)
            if this_checksum != message.Checksum{
                fmt.Println("the checksum is incorrect")
                return 
            }


            fmt.Println("test 1")
            // just put unordered message into the map!!!
            s.m[this_client_id].unordered_queue[this_seq_num] = string(this_payload) // ?? string()

            fmt.Println("test 2")

            // send ack message back to client
            ack_message := NewAck(this_client_id, this_seq_num)
            message_bytes, error := json.Marshal(ack_message)
            if error != nil{
                fmt.Println("error")
                return 
            }
            // s.Write(this_client_id, message_bytes)
            s.connection.WriteToUDP(message_bytes, s.m[this_client_id].client_addr) // modified


            for {
                if val, ok := s.m[this_client_id].unordered_queue[this_seq_num]; ok {
                    out_sequence := this_client_id
                    out_data := val

                    // if time out -> delete map?

                    if entry, ok_2 := s.m[this_client_id]; ok_2 { // because map is not addressable
                       entry.client_sequence_number += 1
                       s.m[this_client_id] = entry
                    }

                    s.queue.PushBack(pair{out_sequence, out_data})

                } else { // means that the sequence number we have currently has not yet arrived (so nothing we can do here)
                    break
                }
            }

        }

        if message.Type == 2{ // ack (do NOT need to do anything for checkpoint?)
            fmt.Println("Type ACK")

        	this_client_id := message.ConnID
        	fmt.Println("Receive ACK message from client")
        	fmt.Println("this_client_id = ", this_client_id)
        	// s.m[this_client_id].server_sequence_number += 1

            if entry, ok := s.m[this_client_id]; ok {
                entry.server_sequence_number += 1
                s.m[this_client_id] = entry
            }

        	// server's sequence number += 1 ?
        	// move sliding window?
        }

        if message.Type == 3{ // cAck
            fmt.Println("Type cACK")

        	this_client_id := message.ConnID
        	fmt.Println("this_client_id = ", this_client_id)
        	fmt.Println("Receive cACK message from client")
        	// s.m[this_client_id].server_sequence_number += 1

            if entry, ok := s.m[this_client_id]; ok {
                entry.server_sequence_number += 1
                s.m[this_client_id] = entry
            }

        	// s.m[this_client_id].client_sequence_number
        	// server's sequence number += 1 ?
        	// move sliding window?
        }
        
    }

}

// probably do not need write_routine, because we can just write it back in the main routine?


/*
LSP's read
will have a received message, but we need to know which client is from, so we need to return client_id

Just taking things out of server's queue 

從queue拿出來，回傳給他
*/
func (s *server) Read() (int, []byte, error) {

	fmt.Println("Read!!!")

	// take out from the server's queue
	
    if s.queue.Len() > 0 {
        message := s.queue.Front() // queue or processed_queue??? return *element

        elem := s.queue.Front().Value

        if foo, ok := elem.(*pair); ok {
               cur_id, _ := foo.pair_id.(int)
               cur_payload, _ := foo.pair_payload.(string)

               s.queue.Remove(message)
               fmt.Println("queue finish")

               return cur_id, []byte(cur_payload), nil
        }
    }
	
    
    // var message pair = s.queue.Front()

    return -1, nil, errors.New("error in Read()")

	// conn_id := message.ConnID
	// payload := message.Payload

	// } // Blocks indefinitely.

	// return -1, nil, errors.New("not yet implemented")
}

func (s *server) Write(connId int, payload []byte) error {
	fmt.Println("Write starts!!!")

	// WriteToUDP ??

    // 用connID找到這個connection的address，然後再傳出去 (建立連線時，要用map記錄對方的addr)
    // seqNum + 1

    addr := s.m[connId].client_addr // added
    // s.connection.WriteTo(payload, addr) // added

    _, error := s.connection.WriteToUDP(payload, addr)
    if error != nil {
        return error
    }

    fmt.Println("Write finishes!!!")

	return nil
}

func (s *server) CloseConn(connId int) error {
	fmt.Println("CloseConn!!!")

	// delete the client from server's map

	return errors.New("not yet implemented")
}

func (s *server) Close() error {
	fmt.Println("Close!!!")

	// stop the server and clean all resources

	return errors.New("not yet implemented")
}