-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Co-Authored-By: Franco Batastini <66926111+Bata340@users.noreply.github.com>
- Loading branch information
1 parent
686deb8
commit 3c74b7b
Showing
31 changed files
with
679 additions
and
473 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| package main | ||
|
|
||
| type PartialSum struct { | ||
| sumOfPrices float32 | ||
| sumOfRows int | ||
| numOfSavers int | ||
| } |
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package filemanager | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
| log "github.com/sirupsen/logrus" | ||
| "os" | ||
| ) | ||
|
|
||
| func MoveFiles(files []string, folderName string) error { | ||
|
|
||
| if _, err := os.Stat(folderName); os.IsNotExist(err) { | ||
| err := os.Mkdir(folderName, os.ModePerm) | ||
| if err != nil && !os.IsExist(err) { | ||
| log.Errorf("FileMover | Error creating directory %v | %v", folderName, err) | ||
| return err | ||
| } | ||
| } | ||
| for _, file := range files { | ||
| err := os.Rename(file, fmt.Sprintf("%v/%v", folderName, file)) | ||
| if err != nil { | ||
| log.Errorf("FileMover | Error moving file to '%v/%v' | %v", folderName, file, err) | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func RenameFile(file string, newName string) error { | ||
| err := os.Rename(file, newName) | ||
| if err != nil { | ||
| log.Errorf("FileRenamer | Error renaming file | %v", err) | ||
| return err | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func DirectoryExists(file string) bool { | ||
| if _, err := os.Stat(file); errors.Is(err, os.ErrNotExist) { | ||
| return false | ||
| } | ||
| return true | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| package getters | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "github.com/brunograssano/Distribuidos-TP1/common/communication" | ||
| dataStructures "github.com/brunograssano/Distribuidos-TP1/common/data_structures" | ||
| "github.com/brunograssano/Distribuidos-TP1/common/filemanager" | ||
| socketsProtocol "github.com/brunograssano/Distribuidos-TP1/common/protocol/sockets" | ||
| "github.com/brunograssano/Distribuidos-TP1/common/serializer" | ||
| "github.com/brunograssano/Distribuidos-TP1/common/utils" | ||
| log "github.com/sirupsen/logrus" | ||
| ) | ||
|
|
||
| type ClientGetter struct { | ||
| config *GetterConfig | ||
| sph *socketsProtocol.SocketProtocolHandler | ||
| clientId string | ||
| stop chan bool | ||
| join chan bool | ||
| } | ||
|
|
||
| func NewClientGetter(clientSocket *communication.TCPSocket, config *GetterConfig, stop chan bool, join chan bool) *ClientGetter { | ||
| return &ClientGetter{ | ||
| config: config, | ||
| sph: socketsProtocol.NewSocketProtocolHandler(clientSocket), | ||
| clientId: "", | ||
| stop: stop, | ||
| join: join, | ||
| } | ||
| } | ||
|
|
||
| func (c *ClientGetter) HandleClientGetter() { | ||
| msg, err := c.sph.Read() | ||
| if err != nil { | ||
| log.Errorf("Client Getter | Error getting first message with client id") | ||
| return | ||
| } | ||
| defer c.sph.Close() | ||
|
|
||
| c.clientId = msg.ClientId | ||
| if filemanager.DirectoryExists(msg.ClientId) { | ||
| c.sendResults() | ||
| } else { | ||
| c.askLaterForResults() | ||
| } | ||
| c.join <- true | ||
| close(c.join) | ||
| } | ||
|
|
||
| // askLaterForResults Tells the client to wait and finishes the connection | ||
| func (c *ClientGetter) askLaterForResults() { | ||
| log.Infof("Client Getter %v | Client asked for results when they are not ready. Answer 'Later'", c.clientId) | ||
| err := c.sph.Write(&dataStructures.Message{ | ||
| TypeMessage: dataStructures.Later, | ||
| DynMaps: make([]*dataStructures.DynamicMap, 0), | ||
| }) | ||
| if err != nil { | ||
| log.Errorf("Client Getter %v | Error trying to send 'Later' Message to socket...", c.clientId) | ||
| } | ||
| } | ||
|
|
||
| // sendResults Sends the saved results to the client | ||
| func (c *ClientGetter) sendResults() { | ||
| log.Infof("Client Getter %v | Sending results to client", c.clientId) | ||
| var currBatch []*dataStructures.DynamicMap | ||
| curLengthOfBatch := 0 | ||
| for _, filename := range c.config.FileNames { | ||
| reader, err := filemanager.NewFileReader(fmt.Sprintf("%v/%v_%v.csv", c.clientId, filename, c.clientId)) | ||
| if err != nil { | ||
| log.Errorf("Client Getter %v | Error trying to open file: %v | %v | Skipping it...", c.clientId, filename, err) | ||
| continue | ||
| } | ||
| for reader.CanRead() { | ||
| select { | ||
| case <-c.stop: | ||
| log.Warnf("Client Getter %v | Received signal while sending file, stopping transfer", c.clientId) | ||
| return | ||
| default: | ||
| } | ||
| line := reader.ReadLine() | ||
| currBatch = append(currBatch, serializer.DeserializeFromString(line)) | ||
| curLengthOfBatch++ | ||
| if uint(curLengthOfBatch) >= c.config.MaxLinesPerSend { | ||
| c.sendBatch(currBatch) | ||
| currBatch = make([]*dataStructures.DynamicMap, 0) | ||
| } | ||
| } | ||
| err = reader.Err() | ||
| if err != nil { | ||
| log.Errorf("Client Getter %v| Error reading file | %v", c.clientId, err) | ||
| } | ||
| utils.CloseFileAndNotifyError(reader.FileManager) | ||
| } | ||
| if curLengthOfBatch > 0 { | ||
| c.sendBatch(currBatch) | ||
| } | ||
| c.sendEOF() | ||
| } | ||
|
|
||
| func (c *ClientGetter) sendEOF() { | ||
| log.Infof("Client Getter %v | Sending EOF to client...", c.clientId) | ||
| err := c.sph.Write(&dataStructures.Message{ | ||
| TypeMessage: dataStructures.EOFGetter, | ||
| DynMaps: []*dataStructures.DynamicMap{}, | ||
| }) | ||
| if err != nil { | ||
| log.Errorf("Client Getter %v | Error trying to send EOF | %v", c.clientId, err) | ||
| } | ||
| } | ||
|
|
||
| // sendBatch sends a specific batch of DynamicMaps via protocol handler | ||
| func (c *ClientGetter) sendBatch(batch []*dataStructures.DynamicMap) { | ||
| err := c.sph.Write(&dataStructures.Message{ | ||
| TypeMessage: dataStructures.FlightRows, | ||
| DynMaps: batch, | ||
| }) | ||
| if err != nil { | ||
| log.Errorf("Client Getter %v | Error sending batch from getter | %v", c.clientId, err) | ||
| } | ||
| } |
Oops, something went wrong.