feat: initial source implementation #2
Conversation
source/iterator.go
Outdated
| metadata := iter.createMetadata(fileInfo, fullPath, len(chunk)) | ||
| metadata["chunk_index"] = fmt.Sprintf("%d", chunkIndex) | ||
| metadata["total_chunks"] = fmt.Sprintf("%d", totalChunks) | ||
| metadata["hash"] = hash(fileInfo.modTime.Format(time.RFC3339)) |
There was a problem hiding this comment.
Nitpick: so that a user doesn't think this is the chunk content hash, it would be good to change the name to something like file_mod_time_hash, or even just the time?
There was a problem hiding this comment.
I've been thinking a bit about the hash. Every file with the same modification time will have the same hash, and that's quite possible. Using the file name, creation time, and/or last modification time together might be an alternative. Two files with the same path might not be the same file, because a file might have been created, then deleted, then created again with the same name.
There was a problem hiding this comment.
Yes, now I have used filename, mod time and filesize to create the hash.
source/source.go
Outdated
| s.ch = make(chan opencdc.Record) | ||
| s.wg = &sync.WaitGroup{} | ||
|
|
||
| s.wg.Add(1) | ||
| err = NewIterator(ctx, s.sshClient, s.sftpClient, s.position, s.config, s.ch, s.wg) |
There was a problem hiding this comment.
Now that I'm reading this code again, why do we actually need a channel that's shared between the source and the iterator? Can the iterator return a record on demand, i.e. when the source's Read() method is called, it calls iterator.Next(), and gets the next record or chunk?
There was a problem hiding this comment.
Yes, now I have refactored the iterator to return records on demand.
source/iterator.go
Outdated
| fullPath := filepath.Join(iter.config.DirectoryPath, filename) | ||
|
|
||
| // Get initial file stat. | ||
| initialStat, err := iter.sftpClient.Stat(fullPath) |
There was a problem hiding this comment.
IIUC, the source works this way: we get a list of file paths, we go through the list, and get the files.
If that's the case, then modifications to files can happen between the time when we get the list of files and when we fetch a file. A file can be moved/deleted, and in that case, I don't think we should fail.
There was a problem hiding this comment.
Yes this could happen, handled it now.
Source implementation for SFTP connector.
Quick checks: