aboutsummaryrefslogtreecommitdiffstats
path: root/Software/Visual_Studio/SideChains/RealTimeGraphEx/Controllers/GraphMultiController.cs
blob: 68fde983b416a67e568a63e3d39c7c3dbb89f7f9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
using RealTimeGraphEx.DataSeries;
using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows;

namespace RealTimeGraphEx.Controllers
{
    public class GraphMultiController : GraphControllerBase
    {
        private readonly object _lock = new object();

        #region Cross Thread Fields

        internal ObservableCollection<DataYSeries> dataSeriesCollection;

        #endregion

        public int TotalPoints
        {
            get { return dataSeriesCollection.Count > 0 ? dataSeriesCollection[0].Points.Count : 0; }
        }

        internal bool IsAllSeriesEqual
        {
            get { return dataSeriesCollection.All(x => x.Points.Count == dataSeriesCollection[0].Points.Count); }
        }

        internal void TrimToLast()
        {
            lock (dataSeriesCollection)
            {
                if (dataSeriesCollection.Count > 1)
                {
                    var min = dataSeriesCollection.FirstOrDefault(x => x.Points.Count == dataSeriesCollection.Min(y => y.Points.Count));

                    if (min != null)
                    {
                        lock (min.Points)
                        {
                            foreach (var series in dataSeriesCollection)
                            {
                                lock (series.Points)
                                {
                                    if (series.Points.Count > min.Points.Count)
                                    {
                                        series.Points.RemoveRange(min.Points.Count, series.Points.Count - min.Points.Count);
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }

        public ObservableCollection<DataYSeries> DataSeriesCollection
        {
            get { return (ObservableCollection<DataSeries.DataYSeries>)GetValue(DataSeriesCollectionProperty); }
            set { SetValue(DataSeriesCollectionProperty, value); }
        }
        public static readonly DependencyProperty DataSeriesCollectionProperty =
            DependencyProperty.Register("Controller", typeof(ObservableCollection<DataYSeries>), typeof(GraphMultiController), new PropertyMetadata(null, new PropertyChangedCallback(SetCrossThreadFields)));

        private static void SetCrossThreadFields(DependencyObject d, DependencyPropertyChangedEventArgs e)
        {
            (d as GraphMultiController).OnSetCrossModelFields();
        }

        protected virtual void OnSetCrossModelFields()
        {
            this.Dispatcher.Invoke(() =>
            {

                dataSeriesCollection = DataSeriesCollection;

            }, System.Windows.Threading.DispatcherPriority.Send);
        }

        public GraphMultiController()
            : base()
        {
            DataSeriesCollection = new ObservableCollection<DataSeries.DataYSeries>();
        }

        public override void Clear()
        {
            if (clearMethod != null)
            {
                clearMethod();
            }

            dataSeriesCollection.ToList().ForEach(x => x.ClearPoints());
        }

        internal override void ClearPoints()
        {
            dataSeriesCollection.ToList().ForEach(x => x.ClearPoints());
        }

        public virtual void PushData(List<double> values)
        {
            lock (_lock)
            {
                if (startThreadMethod != null)
                {
                    startThreadMethod();
                }

                dataSeriesCollection.ToList().ForEach(x => x.PushData(values));
            }
        }

        public virtual void PushData(double[] values)
        {
            lock (_lock)
            {
                if (startThreadMethod != null)
                {
                    startThreadMethod();
                }

                dataSeriesCollection.ToList().ForEach(x => x.PushData(values));
            }
        }

        public virtual void PushData(params List<double>[] arrays)
        {
            PushData(arrays.ToList());
        }

        public virtual void PushData(List<List<double>> arrays)
        {
            lock (_lock)
            {
                if (startThreadMethod != null)
                {
                    startThreadMethod();
                }

                for (int i = 0; i < arrays.Count; i++)
                {
                    dataSeriesCollection[i].PushData(arrays[i]);
                }
            }
        }

        public virtual void PushData(List<double> array, int dataSeriesIndex)
        {
            lock (_lock)
            {
                if (startThreadMethod != null)
                {
                    startThreadMethod();
                }

                dataSeriesCollection[dataSeriesIndex].PushData(array);
            }
        }

        public virtual void PushDataSingles(List<double> values)
        {
            lock (_lock)
            {
                if (startThreadMethod != null)
                {
                    startThreadMethod();
                }

                for (int i = 0; i < values.Count; i++)
                {
                    dataSeriesCollection[i].PushData(values[i]);
                }
            }
        }

        internal virtual List<List<double>> GetAndClearAllPoints()
        {
            List<List<double>> pointsCollection = new List<List<double>>();

            lock (_lock)
            {
                foreach (var series in dataSeriesCollection)
                {
                    pointsCollection.Add(series.Points.GetAndClearAllPoints());
                }
            }

            return pointsCollection;
        }

        public void AddSeries(DataYSeries dataSeries)
        {
            DataSeriesCollection.Add(dataSeries);
        }
    }
}
leSystem.Network; using Tango.FSE.Common.Authentication; using Tango.FSE.Common.Build; using Tango.FSE.Common.Connection; using Tango.FSE.Common.FileSystem; using Tango.FSE.Common.Notifications; using Tango.Transport; using Tango.Transport.Transporters; using Tango.WebRTC; namespace Tango.FSE.UI.FileSystem { /// <summary> /// Represents the <see cref="IFileSystemProvider"/> default implementation. /// </summary> /// <seealso cref="Tango.Core.ExtendedObject" /> /// <seealso cref="Tango.FSE.Common.FileSystem.IFileSystemProvider" /> public class DefaultFileSystemProvider : ExtendedObject, IFileSystemProvider { private IMachineProvider _machineProvider; private BasicTransporter _webRtcTransporter; private const string WEB_RTC_CHANNEL_NAME = "FileSystemChannel"; private const long MAX_CHUNK_SIZE = 1024 * 100; //Max chunk size for the standard channel (WebSockets/SignalR). private const long MIN_CHUNK_SIZE = 1024; //Min chunk size for the standard channel. private const long MAX_CHUNK_SIZE_WEB_RTC = 1024 * 50; //Max chunk size for the WebRTC channel. private const long MIN_CHUNK_SIZE_WEB_RTC = 1024; //Min chunk size for the WebRTC channel. private const int WEB_RTC_MAX_RETRIES = 8; //Maximum number of retries per chunk for the WebRTC channel until falling back to standard channel. private List<FileSystemHandler> _activeHandlers; [TangoInject] private INotificationProvider NotificationProvider { get; set; } [TangoInject] private IAuthenticationProvider AuthenticationProvider { get; set; } [TangoInject(Mode = TangoInjectMode.WhenAvailable)] private IBuildProvider BuildProvider { get; set; } #region Properties private bool _enableWebRTC; /// <summary> /// Gets or sets a value indicating whether to enable a P2P WebRTC channel for fast transport. /// </summary> public bool EnableWebRTC { get { return _enableWebRTC; } set { _enableWebRTC = value; RaisePropertyChangedAuto(); if (value) { LogManager.Log("File system WebRTC channel is now enabled."); } else { LogManager.Log("File system WebRTC channel is now disabled."); } } } private bool _isWebRtcAvailable; /// <summary> /// Gets a value indicating whether the WebRTC channel is available. /// </summary> public bool IsWebRtcAvailable { get { return _isWebRtcAvailable; } private set { _isWebRtcAvailable = value; RaisePropertyChangedAuto(); if (value) { LogManager.Log("File system WebRTC channel is now available."); } else { LogManager.Log("File system WebRTC channel is now available."); } } } #endregion #region Constructors /// <summary> /// Initializes a new instance of the <see cref="DefaultFileSystemProvider"/> class. /// </summary> /// <param name="machineProvider">The machine provider.</param> public DefaultFileSystemProvider(IMachineProvider machineProvider) { _activeHandlers = new List<FileSystemHandler>(); EnableWebRTC = true; //TODO: From Settings.. _machineProvider = machineProvider; _machineProvider.MachineConnected += _machineProvider_MachineConnected; _machineProvider.MachineDisconnected += _machineProvider_MachineDisconnected; } #endregion #region Machine Connection Event Handlers private async void _machineProvider_MachineDisconnected(object sender, MachineDisconnectedEventArgs e) { IsWebRtcAvailable = false; foreach (var handler in _activeHandlers.ToList()) { try { handler.RaiseFailed(new TransporterDisconnectedException("Machine disconnected.")); } catch (Exception ex) { Debug.WriteLine(ex); } } _activeHandlers.Clear(); if (_webRtcTransporter != null) { try { LogManager.Log("Machine disconnected. Disconnecting FileSystem WebRTC Transporter..."); await _webRtcTransporter.Disconnect(); } catch (Exception ex) { LogManager.Log(ex, "Error while disconnecting FileSystem WebRTC Transporter."); } } } private async void _machineProvider_MachineConnected(object sender, MachineConnectedEventArgs e) { if (_machineProvider.ConnectionType.IsRemote()) { try { IsWebRtcAvailable = false; await _machineProvider.MachineOperator.SendGenericRequest<InitWebRtcRequest, InitWebRtcResponse>(new InitWebRtcRequest() { DataChannelName = WEB_RTC_CHANNEL_NAME }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(60), Priority = QueuePriority.Low }); _webRtcTransporter = new BasicTransporter(new WebRtcTransportAdapter(_machineProvider.MachineOperator, WebRtcTransportAdapterMode.Active, WEB_RTC_CHANNEL_NAME) { EnableCompression = _machineProvider.MachineOperator.Adapter.EnableCompression, }); _webRtcTransporter.UseKeepAlive = false; _webRtcTransporter.GenericProtocol = _machineProvider.MachineOperator.GenericProtocol; _webRtcTransporter.ComponentName = "File System Active WebRTC Transporter"; await _webRtcTransporter.Connect(); IsWebRtcAvailable = true; LogManager.Log("FileSystem via WebRTC is ready."); } catch (Exception ex) { IsWebRtcAvailable = false; LogManager.Log(ex, "Error initializing FileSystem via WebRTC."); NotificationProvider.PushErrorReportingSnackbar(ex, "PPC Module Warning", "Could not initialize the remote file system P2P channel.", MessageType.Warning); } } } #endregion #region Public Methods /// <summary> /// Gets a folder by the specified path. /// </summary> /// <param name="path">The path.</param> /// <returns></returns> public async Task<IFileSystemContainer> GetFolder(string path) { try { LogManager.Log($"Retrieving remote folder '{path}'..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<GetFileSystemItemRequest, GetFileSystemItemResponse>(new GetFileSystemItemRequest() { Path = path, }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(30), }); return FileSystemItem.FromDTO(response.FileSystemItem) as IFileSystemContainer; } catch (Exception ex) { throw LogManager.Log(ex, "Error retrieving remote folder."); } } /// <summary> /// Gets the specified special folder. /// </summary> /// <param name="specialFolder">The special folder.</param> /// <returns></returns> public async Task<IFileSystemContainer> GetSpecialFolder(Environment.SpecialFolder specialFolder) { try { LogManager.Log($"Retrieving remote special folder '{specialFolder}'..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<GetFileSystemItemRequest, GetFileSystemItemResponse>(new GetFileSystemItemRequest() { SpecialFolder = specialFolder }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(30), }); return FileSystemItem.FromDTO(response.FileSystemItem) as IFileSystemContainer; } catch (Exception ex) { throw LogManager.Log(ex, "Error retrieving remote special folder."); } } /// <summary> /// Gets the ThisPC (the root path of the PC). /// </summary> /// <returns></returns> public async Task<IFileSystemContainer> GetThisPC() { try { LogManager.Log("Retrieving remote root directory (This PC)..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<GetFileSystemItemRequest, GetFileSystemItemResponse>(new GetFileSystemItemRequest() { //No parameters at all }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(30), }); return FileSystemItem.FromDTO(response.FileSystemItem) as IFileSystemContainer; } catch (Exception ex) { throw LogManager.Log(ex, "Error retrieving remote root directory."); } } /// <summary> /// Downloads the specified file or folder item. /// </summary> /// <param name="remotePath">The remote file or folder.</param> /// <param name="isRemotePathFile">Indicates whether the remote path is a file.</param> /// <param name="localTargetFolderOrFile">The local target folder or file.</param> /// <param name="isLocalTargetFile">Indicates whether the localTargetFolder is a file.</param> /// <returns></returns> public Task<FileSystemHandler> Download(FileSystemItem item, String localTargetFolderOrFile, bool isLocalTargetFile = false) { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemRead); LogManager.Log($"Downloading remote item '{item.Path}' to local path '{localTargetFolderOrFile}'..."); String operationId = String.Empty; String destination = String.Empty; long downloadLength = 0; bool aborted = false; FileSystemHandler handler = null; if (isLocalTargetFile) { destination = localTargetFolderOrFile; } else { destination = Path.Combine(localTargetFolderOrFile, item.Name); } handler = new FileSystemHandler(item.Type == FileSystemItemType.Folder ? FileSystemHandlerType.FolderDownload : FileSystemHandlerType.FileDownload, item, destination, async () => { if (!aborted) { LogManager.Log($"Download aborted by user for '{item.Name}'. Aborting download..."); aborted = true; try { LogManager.Log("Sending abort download operation request..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<AbortOperationRequest, AbortOperationResponse>( new AbortOperationRequest() { OperationId = operationId }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(30) }); LogManager.Log($"Download operation for '{item.Name} 'aborted successfully."); } catch (Exception ex) { LogManager.Log(ex, $"Error aborting the download operation for '{item.Name}'."); } finally { handler.RaiseAborted(); } } }); _activeHandlers.Add(handler); ThreadFactory.StartNew(async () => { try { if (item.Type == FileSystemItemType.File) { LogManager.Log("Download item identified as a file. Sending file download request..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<FileDownloadRequest, FileDownloadResponse>( new FileDownloadRequest() { Path = item.Path }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(20) }); LogManager.Log($"File download response received:\n{response.ToJsonString()}"); operationId = response.OperationId; downloadLength = response.Length; handler.OperationId = operationId; } else if (item.Type == FileSystemItemType.Folder) { LogManager.Log("Download item identified as a folder. Sending folder download request..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<FolderDownloadRequest, FolderDownloadResponse>( new FolderDownloadRequest() { Path = item.Path }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(60) }); LogManager.Log($"folder download response received:\n{response.ToJsonString()}"); operationId = response.OperationId; downloadLength = response.Length; handler.OperationId = operationId; } else { throw new NotSupportedException("The requested file system item is not supported for downloading."); } } catch (Exception ex) { LogManager.Log(ex, $"Error downloading remote item '{item.Path}'."); _activeHandlers.Remove(handler); handler.RaiseFailed(ex); return; } long position = 0; bool webRtcFailed = false; int webRtcRetries = WEB_RTC_MAX_RETRIES; long dynamicMaxChunkSizeSignalR = MAX_CHUNK_SIZE; long dynamicMaxChunkSizeWebRTC = MIN_CHUNK_SIZE_WEB_RTC; bool isWebRTCChunkSizeFixed = false; var tempFile = TemporaryManager.CreateFile(); LogManager.Log($"Generated temporary local file '{tempFile}'..."); LogManager.Log("Starting chunks download..."); LogManager.Log($"WebRTC active: {(IsWebRtcAvailable && EnableWebRTC).ToStringYesNo()}."); while (position < downloadLength && !aborted) { if (handler.IsPaused) { Thread.Sleep(1000); continue; } try { ChunkDownloadResponse response = null; ChunkDownloadRequest request = new ChunkDownloadRequest() { MaxChunkSize = MAX_CHUNK_SIZE, OperationId = operationId, Position = position, }; if (_webRtcTransporter != null && _webRtcTransporter.State == TransportComponentState.Connected && EnableWebRTC && !webRtcFailed && _machineProvider.ConnectionType == MachineConnectionTypes.SignalR) { try { request.MaxChunkSize = dynamicMaxChunkSizeWebRTC; response = await _webRtcTransporter.SendGenericRequest<ChunkDownloadRequest, ChunkDownloadResponse>(request, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(2), Priority = QueuePriority.Low }); if (!isWebRTCChunkSizeFixed) { dynamicMaxChunkSizeWebRTC = Math.Min(dynamicMaxChunkSizeWebRTC + 1024, MAX_CHUNK_SIZE_WEB_RTC); } webRtcRetries = WEB_RTC_MAX_RETRIES; } catch (Exception ex) { webRtcRetries--; dynamicMaxChunkSizeWebRTC = Math.Max(dynamicMaxChunkSizeWebRTC - 1024, MIN_CHUNK_SIZE_WEB_RTC); isWebRTCChunkSizeFixed = true; if (webRtcRetries == 0) { webRtcFailed = true; LogManager.Log(ex, $"WebRTC chunk download for '{item.Name}' failed after {WEB_RTC_MAX_RETRIES} retries with exception:\n{ex.FlattenMessage()}\nFalling back to SignalR download..."); } continue; } } else { request.MaxChunkSize = dynamicMaxChunkSizeSignalR; Stopwatch watch = new Stopwatch(); watch.Start(); response = await _machineProvider.MachineOperator.SendGenericRequest<ChunkDownloadRequest, ChunkDownloadResponse>(request, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(30), Priority = QueuePriority.Low }); watch.Stop(); if (watch.Elapsed.TotalSeconds < 1) { dynamicMaxChunkSizeSignalR += 1024 * 10; } else if (watch.Elapsed.TotalSeconds > 1) { dynamicMaxChunkSizeSignalR -= 1024 * 10; } dynamicMaxChunkSizeSignalR = Math.Max(dynamicMaxChunkSizeSignalR, MIN_CHUNK_SIZE); } using (FileStream fs = new FileStream(tempFile, FileMode.Append)) { fs.Write(response.Data, 0, response.Data.Length); } position += response.Data.Length; handler.InvalidateProgress(position, downloadLength); } catch (Exception ex) { LogManager.Log(ex, $"Download failed for '{item.Name}'."); _activeHandlers.Remove(handler); tempFile.Delete(); handler.RaiseFailed(ex); return; } } if (!aborted) { LogManager.Log($"Chunks completed for '{item.Name}'."); try { if (item.Type == FileSystemItemType.File) { LogManager.Log("Copying temporary file to destination directory..."); File.Copy(tempFile, destination, true); LogManager.Log("Removing temporary file..."); tempFile.Delete(); } else if (item.Type == FileSystemItemType.Folder) { LogManager.Log("Extracting temporary zip file to destination directory..."); using (Ionic.Zip.ZipFile zip = new Ionic.Zip.ZipFile(tempFile)) { zip.ExtractAll(destination, Ionic.Zip.ExtractExistingFileAction.OverwriteSilently); } LogManager.Log("Removing temporary zip file..."); tempFile.Delete(); } LogManager.Log($"Download completed successfully for '{item.Name}'."); handler.RaiseCompleted(); } catch (Exception ex) { LogManager.Log(ex, $"Download failed on the final stage for {item.Name}."); handler.RaiseFailed(ex); } } else { tempFile.Delete(); } _activeHandlers.Remove(handler); }); return Task.FromResult(handler); } /// <summary> /// Downloads the specified file or folder item. /// </summary> /// <param name="remotePath">The remote file or folder.</param> /// <param name="isRemotePathFile">Indicates whether the remote path is a file.</param> /// <param name="localTargetFolderOrFile">The local target folder or file.</param> /// <param name="isLocalTargetFile">Indicates whether the localTargetFolder is a file.</param> /// <returns></returns> public Task<FileSystemHandler> Download(String remotePath, bool isRemotePathFile, String localTargetFolderOrFile, bool isLocalTargetFile = false) { FileSystemItem item = null; if (isRemotePathFile) { item = new FileItem(); } else { item = new FolderItem(); } item.Path = remotePath; return Download(item, localTargetFolderOrFile, isLocalTargetFile); } /// <summary> /// Uploads the specified local file or folder. /// </summary> /// <param name="localSourcePath">The local source path.</param> /// <param name="remoteFolder">The remote folder.</param> /// <returns></returns> /// <exception cref="System.IO.FileNotFoundException">Could not locate the local file or directory to upload.</exception> public Task<FileSystemHandler> Upload(String localSourcePath, FileSystemItem remoteFolder) { String destination = Path.Combine(remoteFolder.Path, Path.GetFileName(localSourcePath)); return Upload(localSourcePath, destination); } /// <summary> /// Uploads the specified local file or folder. /// </summary> /// <param name="localSourcePath">The local source path.</param> /// <param name="remotePath">The remote destination path.</param> /// <param name="forRemoteUpgrade">Indicates whether this upload operation is performed for a remote upgrade.</param> /// <returns></returns> /// <exception cref="FileNotFoundException">Could not locate the local file or directory to upload.</exception> public Task<FileSystemHandler> Upload(String localSourcePath, String remotePath, bool forRemoteUpgrade = false) { if (!BuildProvider.IsTwineRSM) { if (!forRemoteUpgrade) { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemWrite); } else { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_RemoteUpgradeOnline); } } String operationId = String.Empty; String destination = remotePath; bool isFolder = false; bool aborted = false; FileSystemItem sourceItem = null; if (Directory.Exists(localSourcePath)) { LogManager.Log($"Uploading local folder '{localSourcePath}' to remote path '{destination}'..."); sourceItem = new FolderItem() { Path = localSourcePath }; isFolder = true; } else if (File.Exists(localSourcePath)) { LogManager.Log($"Uploading local file '{localSourcePath}' to remote path '{destination}'..."); sourceItem = new FileItem() { Path = localSourcePath }; isFolder = false; } else { throw LogManager.Log(new FileNotFoundException("Could not locate the local file or directory to upload."), "Error uploading local item to remote location."); } FileSystemHandler handler = null; handler = new FileSystemHandler(isFolder ? FileSystemHandlerType.FolderUpload : FileSystemHandlerType.FileUpload, sourceItem, destination, async () => { if (!aborted) { LogManager.Log($"Upload aborted by user for item '{sourceItem.Name}'. Aborting upload..."); aborted = true; try { LogManager.Log("Sending upload operation abort request..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<AbortOperationRequest, AbortOperationResponse>( new AbortOperationRequest() { OperationId = operationId }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(30) }); LogManager.Log($"Upload operation successfully aborted for item '{sourceItem.Name}'."); } catch (Exception ex) { LogManager.Log(ex, $"Error aborting the upload operation for item '{sourceItem.Name}'."); } finally { handler.RaiseAborted(); } } }); _activeHandlers.Add(handler); ThreadFactory.StartNew(async () => { try { if (!isFolder) { LogManager.Log("Upload item identified as a file. Sending file upload request..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<FileUploadRequest, FileUploadResponse>( new FileUploadRequest() { Path = destination }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(20) }); LogManager.Log($"File upload response received:\n{response.ToJsonString()}"); operationId = response.OperationId; handler.OperationId = operationId; } else { LogManager.Log("Upload item identified as a folder. Sending file upload request..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<FolderUploadRequest, FolderUploadResponse>( new FolderUploadRequest() { Path = destination }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(20) }); LogManager.Log($"Folder upload response received:\n{response.ToJsonString()}"); operationId = response.OperationId; handler.OperationId = operationId; } if (isFolder) { var originalPath = localSourcePath; localSourcePath = TemporaryManager.CreateImaginaryFile().Path; LogManager.Log($"Compressing local folder to temporary zip file '{localSourcePath}'..."); ZipFile.CreateFromDirectory(originalPath, localSourcePath); } long position = 0; bool webRtcFailed = false; int webRtcRetries = WEB_RTC_MAX_RETRIES; long dynamixMaxChunkSizeSignalR = MAX_CHUNK_SIZE; long dynamicMaxChunkSizeWebRTC = MIN_CHUNK_SIZE_WEB_RTC; bool isWebRTCChunkSizeFixed = false; LogManager.Log("Starting chunks download..."); LogManager.Log($"WebRTC active: {(IsWebRtcAvailable && EnableWebRTC).ToStringYesNo()}."); using (FileStream fs = new FileStream(localSourcePath, FileMode.Open)) { while (position < fs.Length && !aborted) { fs.Position = position; if (handler.IsPaused) { Thread.Sleep(1000); continue; } try { ChunkUploadResponse response = null; ChunkUploadRequest request = new ChunkUploadRequest() { OperationId = operationId, }; if (_webRtcTransporter != null && _webRtcTransporter.State == TransportComponentState.Connected && EnableWebRTC && !webRtcFailed && _machineProvider.ConnectionType == MachineConnectionTypes.SignalR) { try { byte[] data = new byte[Math.Min(dynamicMaxChunkSizeWebRTC, fs.Length - fs.Position)]; fs.Read(data, 0, data.Length); request.Data = data; request.IsCompleted = fs.Position == fs.Length; response = await _webRtcTransporter.SendGenericRequest<ChunkUploadRequest, ChunkUploadResponse>(request, new TransportRequestConfig() { Timeout = request.IsCompleted ? TimeSpan.FromSeconds(120) : TimeSpan.FromSeconds(2), Priority = QueuePriority.Low }); if (!isWebRTCChunkSizeFixed) { dynamicMaxChunkSizeWebRTC = Math.Min(dynamicMaxChunkSizeWebRTC + 1024, MAX_CHUNK_SIZE_WEB_RTC); } webRtcRetries = WEB_RTC_MAX_RETRIES; } catch (Exception ex) { webRtcRetries--; dynamicMaxChunkSizeWebRTC = Math.Max(dynamicMaxChunkSizeWebRTC - 1024, MIN_CHUNK_SIZE_WEB_RTC); isWebRTCChunkSizeFixed = true; if (webRtcRetries == 0) { webRtcFailed = true; LogManager.Log(ex, $"WebRTC chunk upload for '{sourceItem.Name}' failed after {WEB_RTC_MAX_RETRIES} retries with exception:\n{ex.FlattenMessage()}\nFalling back to SignalR upload..."); } continue; } } else { byte[] data = new byte[Math.Min(dynamixMaxChunkSizeSignalR, fs.Length - fs.Position)]; fs.Read(data, 0, data.Length); request.Data = data; request.IsCompleted = fs.Position == fs.Length; Stopwatch watch = new Stopwatch(); watch.Start(); response = await _machineProvider.MachineOperator.SendGenericRequest<ChunkUploadRequest, ChunkUploadResponse>(request, new TransportRequestConfig() { Timeout = request.IsCompleted ? TimeSpan.FromSeconds(120) : TimeSpan.FromSeconds(30), Priority = QueuePriority.Low }); watch.Stop(); if (watch.Elapsed.TotalSeconds < 1) { dynamixMaxChunkSizeSignalR += 1024 * 10; } else if (watch.Elapsed.TotalSeconds > 1) { dynamixMaxChunkSizeSignalR -= 1024 * 10; } dynamixMaxChunkSizeSignalR = Math.Max(dynamixMaxChunkSizeSignalR, MIN_CHUNK_SIZE); } position = fs.Position; handler.InvalidateProgress(position, fs.Length); } catch (Exception ex) { LogManager.Log(ex, $"Upload failed for '{sourceItem.Name}'."); _activeHandlers.Remove(handler); handler.RaiseFailed(ex); if (isFolder) { try { fs.Dispose(); LogManager.Log("Removing temporary zip file..."); File.Delete(localSourcePath); } catch { } } return; } } if (!aborted) { LogManager.Log($"Upload for '{sourceItem.Name}' completed successfully."); handler.RaiseCompleted(); } if (isFolder) { try { LogManager.Log("Removing temporary zip file..."); File.Delete(localSourcePath); } catch { } } } } catch (Exception ex) { LogManager.Log(ex, $"Error uploading local item '{sourceItem.Name}'."); _activeHandlers.Remove(handler); handler.RaiseFailed(ex); return; } _activeHandlers.Remove(handler); }); return Task.FromResult(handler); } /// <summary> /// Copies the specified remote file or folder to the specified target remote folder. /// </summary> /// <param name="source">The remote source file or folder.</param> /// <param name="target">The remote target folder.</param> /// <returns></returns> /// <exception cref="System.NotSupportedException"> /// The source file system item is not supported for copying. /// or /// The target file system item is not a valid container. /// </exception> public async Task Copy(FileSystemItem source, FileSystemItem target) { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemWrite); try { LogManager.Log($"Copying remote item '{source.Path}' to '{target.Path}'..."); if (source.Type == FileSystemItemType.Drive) { throw new NotSupportedException("The source file system item is not supported for copying."); } if (target.Type == FileSystemItemType.File) { throw new NotSupportedException("The target file system item is not a valid container."); } await _machineProvider.MachineOperator.SendGenericRequest<CopyRequest, CopyResponse>(new CopyRequest() { Source = source.Path, Destination = Path.Combine(target.Path, source.Name) }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(120), }); } catch (Exception ex) { throw LogManager.Log(ex, "Error performing copy operation on remote file system."); } } /// <summary> /// Moves the specified remote file or folder to the remote target folder. /// </summary> /// <param name="source">The remote source file or folder.</param> /// <param name="target">The remote target folder.</param> /// <returns></returns> /// <exception cref="System.NotSupportedException"> /// The source file system item is not supported for copying. /// or /// The target file system item is not a valid container. /// </exception> public async Task Move(FileSystemItem source, FileSystemItem target) { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemWrite); try { LogManager.Log($"Moving remote item '{source.Path}' to '{target.Path}'..."); if (source.Type == FileSystemItemType.Drive) { throw new NotSupportedException("The source file system item is not supported for moving."); } if (target.Type == FileSystemItemType.File) { throw new NotSupportedException("The target file system item is not a valid container."); } await _machineProvider.MachineOperator.SendGenericRequest<MoveRequest, MoveResponse>(new MoveRequest() { Source = source.Path, Destination = Path.Combine(target.Path, source.Name) }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(120), }); } catch (Exception ex) { throw LogManager.Log(ex, "Error performing move operation on the remote file system item."); } } /// <summary> /// Renames the specified file or folder. /// </summary> /// <param name="source">The remote source file or folder.</param> /// <param name="newName">The new name.</param> /// <returns></returns> /// <exception cref="System.NotSupportedException">The source file system item is not supported for copying.</exception> /// <exception cref="System.ArgumentException">The new name contains invalid characters.</exception> public async Task Rename(FileSystemItem source, string newName) { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemWrite); try { LogManager.Log($"Renaming remote item '{source.Path}' to '{newName}'..."); if (source.Type == FileSystemItemType.Drive) { throw new NotSupportedException("The source file system item is not supported for renaming."); } if (newName.ToList().Exists(x => Path.GetInvalidFileNameChars().Contains(x))) { throw new ArgumentException("The new name contains invalid characters."); } await _machineProvider.MachineOperator.SendGenericRequest<MoveRequest, MoveResponse>(new MoveRequest() { Source = source.Path, Destination = Path.Combine(Path.GetDirectoryName(source.Path), newName) }); } catch (Exception ex) { throw LogManager.Log(ex, "Error performing rename operation on the remote file system item."); } } /// <summary> /// Deletes the specified file or folder. /// </summary> /// <param name="item">The remote file or folder.</param> /// <returns></returns> /// <exception cref="System.NotSupportedException">The source file system item is not supported for deletion.</exception> public async Task Delete(FileSystemItem item) { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemWrite); try { LogManager.Log($"Deleting remote item '{item.Path}'..."); if (item.Type == FileSystemItemType.Drive) { throw new NotSupportedException("The source file system item is not supported for deletion."); } await _machineProvider.MachineOperator.SendGenericRequest<DeleteRequest, DeleteResponse>(new DeleteRequest() { Path = item.Path }, new TransportRequestConfig() { Timeout = TimeSpan.FromSeconds(120) }); } catch (Exception ex) { throw LogManager.Log(ex, "Error performing delete operation on the remote file system item."); } } /// <summary> /// Creates a new folder at the specified remote parent. /// </summary> /// <param name="parent">The remote parent path.</param> /// <param name="folderName">Name of the new folder.</param> /// <returns></returns> public async Task<FolderItem> CreateFolder(FileSystemItem parent, string folderName) { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemWrite); try { LogManager.Log($"Creating new folder '{folderName}' on the remote path '{parent.Path}'..."); var response = await _machineProvider.MachineOperator.SendGenericRequest<CreateFolderRequest, CreateFolderResponse>(new CreateFolderRequest() { Path = parent.Path, FolderName = folderName, }); return FileSystemItem.FromDTO(response.FolderItem) as FolderItem; } catch (Exception ex) { throw LogManager.Log(ex, "Error creating new folder on the remote file system location."); } } /// <summary> /// Performs a disk space optimization. /// </summary> /// <returns></returns> public async Task<PerformDiskSpaceOptimizationResponse> PerformDiskSpaceOptimization() { AuthenticationProvider.ThrowIfNoPermission(Permissions.FSE_PPCFileSystemRead); try { LogManager.Log("Performing disk space optimization on the remote file system."); var response = await _machineProvider.MachineOperator.SendGenericRequest<PerformDiskSpaceOptimizationRequest, PerformDiskSpaceOptimizationResponse>(new PerformDiskSpaceOptimizationRequest() { }, new TransportRequestConfig() { Timeout = TimeSpan.FromMinutes(5) }); return response; } catch (Exception ex) { throw LogManager.Log(ex, "Error occurred while trying to perform the disk space optimization on the remote file system."); } } #endregion } }