-
-
Notifications
You must be signed in to change notification settings - Fork 346
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
12 changed files
with
387 additions
and
71 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
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,119 @@ | ||
| using System; | ||
| using System.IO; | ||
| using System.Collections.Generic; | ||
| using log4net; | ||
| using CKAN.NetKAN.Model; | ||
| using CKAN.NetKAN.Services; | ||
| using CKAN.NetKAN.Transformers; | ||
| using CKAN.NetKAN.Validators; | ||
|
|
||
| namespace CKAN.NetKAN.Processors | ||
| { | ||
| public class Inflator | ||
| { | ||
| public Inflator(string cacheDir, bool overwriteCache, string githubToken, bool prerelease) | ||
| { | ||
| log.Debug("Initializing inflator"); | ||
| this.githubToken = githubToken; | ||
| this.prerelease = prerelease; | ||
| cache = FindCache( | ||
| new KSPManager(new ConsoleUser(false)), | ||
| new Win32Registry(), | ||
| cacheDir | ||
| ); | ||
| http = new CachingHttpService(cache, overwriteCache); | ||
| ckanValidator = new CkanValidator(http, moduleService); | ||
| } | ||
|
|
||
| internal IEnumerable<Metadata> Inflate(string filename, Metadata netkan, int? releases) | ||
| { | ||
| log.DebugFormat("Inflating {0}", filename); | ||
| try | ||
| { | ||
| // Tell the downloader that we're starting a new request | ||
| http.ClearRequestedURLs(); | ||
|
|
||
| netkanValidator.ValidateNetkan(netkan, filename); | ||
| log.Info("Input successfully passed pre-validation"); | ||
|
|
||
| // TODO: Make this re-usable by refactoring releases somehow | ||
| transformer = new NetkanTransformer( | ||
| http, | ||
| fileService, | ||
| moduleService, | ||
| githubToken, | ||
| prerelease, | ||
| releases | ||
| ); | ||
| IEnumerable<Metadata> ckans = transformer.Transform(netkan); | ||
| log.Info("Finished transformation"); | ||
|
|
||
| foreach (Metadata ckan in ckans) | ||
| { | ||
| ckanValidator.ValidateCkan(ckan, netkan); | ||
| } | ||
| log.Info("Output successfully passed post-validation"); | ||
| return ckans; | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| // Purge anything we download for a failed indexing attempt from the cache to allow re-downloads | ||
| PurgeDownloads(http, cache); | ||
| throw e; | ||
| } | ||
| } | ||
|
|
||
| private static NetFileCache FindCache(KSPManager kspManager, IWin32Registry reg, string cacheDir) | ||
| { | ||
| if (cacheDir != null) | ||
| { | ||
| log.InfoFormat("Using user-supplied cache at {0}", cacheDir); | ||
| return new NetFileCache(cacheDir); | ||
| } | ||
|
|
||
| try | ||
| { | ||
| log.InfoFormat("Using main CKAN meta-cache at {0}", reg.DownloadCacheDir); | ||
| /// Create a new file cache in the same location so NetKAN can download pure URLs not sourced from CkanModules | ||
| return new NetFileCache(kspManager, reg.DownloadCacheDir); | ||
| } | ||
| catch | ||
| { | ||
| // Meh, can't find KSP. 'Scool, bro. | ||
| } | ||
|
|
||
| var tempdir = Path.GetTempPath(); | ||
| log.InfoFormat("Using tempdir for cache: {0}", tempdir); | ||
|
|
||
| return new NetFileCache(tempdir); | ||
| } | ||
|
|
||
| private static void PurgeDownloads(IHttpService http, NetFileCache cache) | ||
| { | ||
| log.Debug("Deleting downloads for failed inflation"); | ||
| if (http != null && cache != null) | ||
| { | ||
| foreach (Uri url in http.RequestedURLs) | ||
| { | ||
| cache.Remove(url); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private string githubToken; | ||
| private bool prerelease; | ||
|
|
||
| private NetFileCache cache; | ||
| private IHttpService http; | ||
|
|
||
| private IModuleService moduleService = new ModuleService(); | ||
| private IFileService fileService = new FileService(); | ||
|
|
||
| private NetkanTransformer transformer; | ||
|
|
||
| private NetkanValidator netkanValidator = new NetkanValidator(); | ||
| private CkanValidator ckanValidator; | ||
|
|
||
| private static readonly ILog log = LogManager.GetLogger(typeof(Inflator)); | ||
| } | ||
| } |
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,205 @@ | ||
| using System; | ||
| using System.IO; | ||
| using System.Text; | ||
| using System.Linq; | ||
| using System.Collections.Generic; | ||
| using System.Globalization; | ||
| using Newtonsoft.Json; | ||
| using Newtonsoft.Json.Linq; | ||
| using Amazon; | ||
| using Amazon.SQS; | ||
| using Amazon.SQS.Model; | ||
| using log4net; | ||
| using CKAN.NetKAN.Model; | ||
|
|
||
| namespace CKAN.NetKAN.Processors | ||
| { | ||
| public class QueueHandler | ||
| { | ||
| public QueueHandler(string cacheDir, bool overwriteCache, string githubToken, bool prerelease) | ||
| { | ||
| log.Debug("Initializing SQS queue handler"); | ||
| // TODO: Shouldn't hard code the region | ||
| client = new AmazonSQSClient(RegionEndpoint.USWest2); | ||
| inflator = new Inflator(cacheDir, overwriteCache, githubToken, prerelease); | ||
|
|
||
| inputQueueURL = getQueueUrl(inputQueueName); | ||
| outputQueueURL = getQueueUrl(outputQueueName); | ||
| log.DebugFormat("Queue URLs: {0}, {1}", inputQueueURL, outputQueueURL); | ||
| } | ||
|
|
||
| public void Process() | ||
| { | ||
| while (true) | ||
| { | ||
| int releases; | ||
| Metadata netkan = getNetkan(out releases); | ||
| if (netkan == null) | ||
| { | ||
| continue; | ||
| } | ||
| log.InfoFormat("Inflating {0}", netkan.Identifier); | ||
| IEnumerable<Metadata> ckans = null; | ||
| try | ||
| { | ||
| ckans = inflator.Inflate($"{netkan.Identifier}.netkan", netkan, releases); | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| e = e.GetBaseException() ?? e; | ||
| log.InfoFormat("Sending failure: {0}", e.Message); | ||
| sendCkan(null, netkan, false, e.Message); | ||
| } | ||
| if (ckans != null) | ||
| { | ||
| foreach (Metadata ckan in ckans) | ||
| { | ||
| log.InfoFormat("Sending {0}-{1}", ckan.Identifier, ckan.Version); | ||
| sendCkan(ckan, netkan, true); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private string getQueueUrl(string name) | ||
| { | ||
| log.DebugFormat("Looking up URL for queue {0}", name); | ||
| return client.GetQueueUrl(new GetQueueUrlRequest() { QueueName = name }).QueueUrl; | ||
| } | ||
|
|
||
| private Metadata getNetkan(out int releases) | ||
| { | ||
| log.Debug("Retrieving metadata from queue"); | ||
| // TODO: Get this from an attribute | ||
| releases = 1; | ||
| var msg = getFromQueue(inputQueueURL); | ||
| if (msg != null) | ||
| { | ||
| log.DebugFormat("Metadata returned: {0}", msg.Body); | ||
| return new Metadata(JObject.Parse(msg.Body)); | ||
| } | ||
| log.Debug("No metadata in queue"); | ||
| return null; | ||
| } | ||
|
|
||
| private Message getFromQueue(string url) | ||
| { | ||
| log.DebugFormat("Looking for message from {0}", url); | ||
| var resp = client.ReceiveMessage(new ReceiveMessageRequest() | ||
| { | ||
| QueueUrl = url, | ||
| MaxNumberOfMessages = 1, | ||
| VisibilityTimeout = (int)TimeSpan.FromMinutes(15).TotalSeconds, | ||
| }); | ||
| var msg = resp.Messages.FirstOrDefault(); | ||
| if (msg != null) | ||
| { | ||
| log.Debug("Message received"); | ||
| client.DeleteMessage(new DeleteMessageRequest() | ||
| { | ||
| QueueUrl = url, | ||
| ReceiptHandle = msg.ReceiptHandle, | ||
| }); | ||
| } | ||
| return msg; | ||
| } | ||
|
|
||
| private void sendCkan(Metadata ckan, Metadata netkan, bool success, string err = null) | ||
| { | ||
| var attribs = new Dictionary<string, MessageAttributeValue>() | ||
| { | ||
| { | ||
| "ModIdentifier", | ||
| new MessageAttributeValue() | ||
| { | ||
| DataType = "String", | ||
| StringValue = netkan.Identifier | ||
| } | ||
| }, | ||
| { | ||
| "Staged", | ||
| new MessageAttributeValue() | ||
| { | ||
| DataType = "String", | ||
| StringValue = netkan.Staged.ToString() | ||
| } | ||
| }, | ||
| { | ||
| "Success", | ||
| new MessageAttributeValue() | ||
| { | ||
| DataType = "String", | ||
| StringValue = success.ToString() | ||
| } | ||
| }, | ||
| { | ||
| "CheckTime", | ||
| new MessageAttributeValue() | ||
| { | ||
| DataType = "String", | ||
| StringValue = DateTime.UtcNow.ToString("s", CultureInfo.InvariantCulture) | ||
| } | ||
| }, | ||
| }; | ||
| if (!string.IsNullOrEmpty(err)) | ||
| { | ||
| attribs.Add( | ||
| "ErrorMessage", | ||
| new MessageAttributeValue() | ||
| { | ||
| DataType = "String", | ||
| StringValue = err | ||
| } | ||
| ); | ||
| } | ||
|
|
||
| SendMessageRequest msg = new SendMessageRequest() | ||
| { | ||
| QueueUrl = outputQueueURL, | ||
| MessageGroupId = "1", | ||
| MessageDeduplicationId = Path.GetRandomFileName(), | ||
| MessageBody = serializeCkan(ckan), | ||
| MessageAttributes = attribs, | ||
| }; | ||
| try | ||
| { | ||
| var resp = client.SendMessage(msg); | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| log.ErrorFormat("Send failed: {0}\r\n{1}", e.Message, e.StackTrace); | ||
| } | ||
| } | ||
|
|
||
| private string serializeCkan(Metadata ckan) | ||
| { | ||
| if (ckan == null) | ||
| { | ||
| return ""; | ||
| } | ||
| var sw = new StringWriter(new StringBuilder()); | ||
| using (var writer = new JsonTextWriter(sw) | ||
| { | ||
| Formatting = Formatting.Indented, | ||
| Indentation = 4, | ||
| IndentChar = ' ', | ||
| }) | ||
| { | ||
| var serializer = new JsonSerializer(); | ||
| serializer.Serialize(writer, ckan.Json()); | ||
| } | ||
| return sw + Environment.NewLine; | ||
| } | ||
|
|
||
| private Inflator inflator; | ||
| private AmazonSQSClient client; | ||
|
|
||
| private const string inputQueueName = "InboundDev.fifo"; | ||
| private const string outputQueueName = "OutboundDev.fifo"; | ||
|
|
||
| private readonly string inputQueueURL; | ||
| private readonly string outputQueueURL; | ||
|
|
||
| private static readonly ILog log = LogManager.GetLogger(typeof(QueueHandler)); | ||
| } | ||
| } |
Oops, something went wrong.