Ingestion service that fetches data from endpoint at regular intervals CSProjectsEAL/GraBID#12 #1
There are no files selected for viewing
This file contains hidden or 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 hidden or 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 |
|---|---|---|
| @@ -1,18 +1,13 @@ | ||
| FROM microsoft/dotnet:2.1-sdk AS build | ||
| COPY . ./build | ||
| WORKDIR /build/ | ||
| RUN dotnet restore | ||
| RUN dotnet build --no-restore -c Release -o /app | ||
| FROM build AS publish | ||
| RUN dotnet publish --no-restore -c Release -o /app | ||
| FROM microsoft/dotnet:2.1-runtime-alpine | ||
| COPY --from=publish /app /var/app | ||
| COPY entrypoint.sh /var/app | ||
| #COPY wait-for-it.sh /var/app | ||
| #RUN apk add bash | ||
| #CMD ["bash","/var/app/wait-for-it.sh","localhost:5672","--timeout=30","--","/var/app/entrypoint.sh"] |
This file contains hidden or 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 |
|---|---|---|
| @@ -1,23 +1,18 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
| <PropertyGroup> | ||
| <OutputType>Exe</OutputType> | ||
| <TargetFramework>netcoreapp2.1</TargetFramework> | ||
| </PropertyGroup> | ||
| <ItemGroup> | ||
| <None Remove="Dockerfile" /> | ||
| </ItemGroup> | ||
| <ItemGroup> | ||
| <Content Include="Dockerfile" /> | ||
| </ItemGroup> | ||
| <ItemGroup> | ||
| <PackageReference Include="Newtonsoft.Json" Version="11.0.2" /> | ||
| <PackageReference Include="RabbitMQ.Client" Version="5.1.0" /> | ||
| <PackageReference Include="Serilog" Version="2.7.1" /> | ||
| <PackageReference Include="Serilog.Sinks.Console" Version="3.1.1" /> | ||
| </ItemGroup> | ||
| </Project> |
This file contains hidden or 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,127 @@ | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.IO; | ||
| using System.Net.Http; | ||
| using System.Threading.Tasks; | ||
| using Newtonsoft.Json; | ||
| using RabbitMQ.Client; | ||
| using Serilog; | ||
| using Shared; | ||
|
|
||
| namespace Ingestion | ||
| { | ||
| public class IngestionService | ||
| { | ||
| private IDictionary<string,IngestionSource> _sourcesList; | ||
| private const string SourcesConfigRelativePath = "../../../../../config/IngestionSourcesTest.json"; | ||
| private ILogger _logger; | ||
| private static IConnection _conn; | ||
| private static IModel _channel; | ||
|
|
||
| public IngestionService(IDictionary<string, IngestionSource> sources, ILogger logger) | ||
| { | ||
| _logger = logger; | ||
| IConnectionFactory factory = new ConnectionFactory() {HostName = "localhost", UserName = "guest", Password = "guest"}; | ||
| _conn = factory.CreateConnection(); | ||
| _channel = _conn.CreateModel(); | ||
| _sourcesList = sources; | ||
| } | ||
|
|
||
| public IngestionService(ILogger logger) | ||
| { | ||
| _logger = logger; | ||
| _sourcesList = LoadSourcesFromJson(SourcesConfigRelativePath); | ||
| IConnectionFactory factory = new ConnectionFactory() {HostName = "localhost", UserName = "guest", Password = "guest"}; | ||
| _conn = factory.CreateConnection(); | ||
| _channel = _conn.CreateModel(); | ||
| } | ||
|
|
||
| private async Task<string> GetData (string url) | ||
| { | ||
| var data = ""; | ||
| using (var client = new HttpClient()) | ||
| { | ||
| using (var res = await client.GetAsync(url)) | ||
| { | ||
| using (var content = res.Content) | ||
| { | ||
| data = await content.ReadAsStringAsync(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return data; | ||
| } | ||
|
|
||
| public IDictionary<string, IngestionSource> LoadSourcesFromJson(string path) | ||
| { | ||
| IDictionary<string, IngestionSource> sources = new Dictionary<string, IngestionSource>(); | ||
| // deserialize JSON directly from a file | ||
| if (File.Exists(path)) | ||
| { | ||
| string JSONText = File.ReadAllText(path); | ||
| IList<IngestionSource> listsources = JsonConvert.DeserializeObject<IList<IngestionSource>>(JSONText); | ||
| foreach (IngestionSource s in listsources) { | ||
| sources.Add(s.Name, s); | ||
| } | ||
| } | ||
| else | ||
| { | ||
| throw new Exception("Cant find file"); | ||
| } | ||
| return sources; | ||
| } | ||
|
|
||
| public void Ingest(string sourceId) { | ||
|
|
||
| if (!_sourcesList.ContainsKey(sourceId)) throw new Exception("Key was not found, source is not defined"); | ||
|
|
||
| _logger.Information("Starting Ingestor"); | ||
|
|
||
| var data = GetData(_sourcesList[sourceId].ApiUrl).Result; | ||
|
|
||
| ForwardMessageToRabbitMQ(data, _sourcesList[sourceId].ForwardMessageQueue); | ||
|
|
||
| _logger.Information("Stopping Ingestor"); | ||
| } | ||
|
|
||
| public void ForwardMessageToRabbitMQ(string message, string queue) | ||
| { | ||
| using (_conn) | ||
| { | ||
| using (_channel) | ||
| { | ||
| var exchange = "mono.data.received"; | ||
|
|
||
| _channel.QueueDeclare(queue, true, false, false, null); | ||
|
|
||
| _channel.ExchangeDeclare(exchange, "fanout"); | ||
|
|
||
| _channel.QueueBind(queue, exchange, ""); | ||
|
|
||
| var envelope = new Envelope<string>(Guid.NewGuid(), message); | ||
| var envelopedMessage = JsonConvert.SerializeObject(envelope); | ||
|
|
||
| byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes(envelopedMessage); | ||
|
|
||
| _channel.BasicPublish(exchange, "", _channel.CreateBasicProperties(), messageBodyBytes); | ||
| Console.WriteLine(" [x] Sent '{0}':'{1}'", exchange, message); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public string SerializeObjects(object obj) | ||
| { | ||
| return JsonConvert.SerializeObject(obj); | ||
| } | ||
|
|
||
| public void StoreObject(string obj) { | ||
| try { | ||
| File.WriteAllText(SourcesConfigRelativePath, obj); | ||
| } catch(Exception exception) | ||
| { | ||
| Console.WriteLine("Couldnt write to file: " + exception.Message + "Message"); | ||
| } | ||
| } | ||
| } | ||
| } | ||
This file contains hidden or 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,14 @@ | ||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Text; | ||
|
|
||
| namespace Ingestion | ||
| { | ||
| public class IngestionSource | ||
| { | ||
| public string Name { get; set; } | ||
| public string ApiUrl { get; set; } | ||
| public string Credential { get; set; } | ||
| public string ForwardMessageQueue { get; set; } | ||
| } | ||
| } |
This file contains hidden or 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 |
|---|---|---|
| @@ -1,38 +1,40 @@ | ||
| using Newtonsoft.Json; | ||
| using RabbitMQ.Client; | ||
| using Serilog; | ||
| using Shared; | ||
| using System; | ||
| using System.Threading; | ||
|
|
||
| namespace Ingestion | ||
| { | ||
| class Program | ||
| { | ||
| static void Main(string[] args) | ||
| { | ||
| ILogger logger = new LoggerConfiguration().WriteTo.Console().CreateLogger(); | ||
| string sourceID = ""; | ||
| /* | ||
| try { | ||
| sourceID = args[0]; | ||
| } catch (Exception e) { | ||
| logger.Error("Incorrect argument for ingestion dll crontab: " + e.Message); | ||
| } | ||
| */ | ||
| sourceID = "ResellerData"; | ||
|
There was a problem hiding this comment. Could be defined in line 15 as well |
||
| if (sourceID != ""){ | ||
| IngestionService ingestService = new IngestionService(logger); | ||
|
|
||
| try { | ||
| while (true) { | ||
| ingestService.Ingest(sourceID); | ||
| Thread.Sleep(3000); | ||
| } | ||
| } catch(Exception e) | ||
| { | ||
| logger.Error("Error: " + e.Message + "\n" + "Exception trace: " + e.StackTrace); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } | ||
| } | ||
This file contains hidden or 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 |
|---|---|---|
| @@ -1,2 +1 @@ | ||
| */5 * * * * /usr/bin/dotnet /var/app/Ingestion.dll ResellerData >> /activity.log 2>&1 |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Shouldn't the _logger be used in this line?