-
Notifications
You must be signed in to change notification settings - Fork 0
Integration Examples
LoSkroefie edited this page Jan 19, 2025
·
1 revision
This guide provides extensive examples of integrating FLEXON with various frameworks and platforms. Each example includes detailed explanations and best practices.
public class Startup
{
public void ConfigureServices(IServiceCollection services)
{
// Add FLEXON formatters
services.AddControllers()
.AddFlexonFormatters(options =>
{
options.EnableCompression = true;
options.ValidationMode = ValidationMode.Strict;
options.UsePooledBuffers = true;
options.BufferSize = 8192;
});
// Add FLEXON serializer as singleton
services.AddSingleton<IFlexonSerializer>(provider =>
new FlexonSerializer(new FlexonOptions
{
EnableCompression = true,
EnableValidation = true,
UsePooledBuffers = true
}));
}
}[ApiController]
[Route("[controller]")]
public class DataController : ControllerBase
{
private readonly IFlexonSerializer _serializer;
public DataController(IFlexonSerializer serializer)
{
_serializer = serializer;
}
[HttpPost]
[Consumes("application/x-flexon")]
[Produces("application/x-flexon")]
public async Task<IActionResult> Post([FromBody] ComplexData data)
{
// Process data
var result = await ProcessData(data);
return Ok(result);
}
[HttpGet("stream")]
public async Task StreamData()
{
Response.ContentType = "application/x-flexon";
using var writer = new FlexonStreamWriter(Response.Body);
await foreach (var item in GetDataStream())
{
await writer.WriteAsync(item);
await Response.Body.FlushAsync();
}
}
}public class FlexonMiddleware
{
private readonly RequestDelegate _next;
private readonly FlexonOptions _options;
public async Task InvokeAsync(HttpContext context)
{
if (context.Request.ContentType == "application/x-flexon")
{
context.Request.EnableBuffering();
using var reader = new FlexonReader(context.Request.Body);
var data = await reader.ReadAsync<dynamic>();
// Process data
context.Items["flexon_data"] = data;
}
await _next(context);
}
}public class FlexonHub : Hub
{
private readonly IFlexonSerializer _serializer;
public async Task StreamData<T>(IAsyncEnumerable<T> data)
{
await foreach (var item in data)
{
var binary = _serializer.Serialize(item);
await Clients.All.SendAsync("DataReceived", binary);
}
}
}
// Client side
var connection = new HubConnectionBuilder()
.WithUrl("/flexonHub")
.AddFlexonProtocol()
.Build();
connection.On<byte[]>("DataReceived", data =>
{
var item = serializer.Deserialize<DataItem>(data);
// Process item
});public class FlexonValueConverter<T> : ValueConverter<T, byte[]>
{
private static readonly IFlexonSerializer _serializer = new FlexonSerializer();
public FlexonValueConverter() : base(
v => _serializer.Serialize(v),
v => _serializer.Deserialize<T>(v))
{
}
}
public class MyContext : DbContext
{
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
modelBuilder.Entity<Document>()
.Property(d => d.Data)
.HasConversion(new FlexonValueConverter<ComplexData>());
}
}public static class FlexonQueryExtensions
{
public static IQueryable<T> WhereFlexon<T>(
this IQueryable<T> query,
Expression<Func<T, byte[]>> property,
Func<dynamic, bool> predicate)
{
return query.Where(e =>
predicate(_serializer.Deserialize<dynamic>(property.Compile()(e))));
}
}
// Usage
var results = await context.Documents
.WhereFlexon(d => d.Data,
data => data.Type == "important" && data.Priority > 5)
.ToListAsync();public class BulkOperations
{
public async Task BulkInsert<T>(DbContext context, IEnumerable<T> entities)
{
var options = new FlexonOptions { EnableCompression = true };
var serializer = new FlexonSerializer(options);
using var transaction = await context.Database.BeginTransactionAsync();
try
{
foreach (var batch in entities.Chunk(1000))
{
var binary = serializer.Serialize(batch);
await context.Database.ExecuteSqlRawAsync(
"EXEC BulkInsert @TableName, @Data",
new SqlParameter("TableName", typeof(T).Name),
new SqlParameter("Data", binary));
}
await transaction.CommitAsync();
}
catch
{
await transaction.RollbackAsync();
throw;
}
}
}public class FlexonRabbitMQ
{
private readonly IModel _channel;
private readonly IFlexonSerializer _serializer;
public async Task PublishAsync<T>(string exchange, string routingKey, T message)
{
var binary = _serializer.Serialize(message);
var props = _channel.CreateBasicProperties();
props.ContentType = "application/x-flexon";
props.DeliveryMode = 2; // persistent
_channel.BasicPublish(
exchange: exchange,
routingKey: routingKey,
mandatory: true,
basicProperties: props,
body: binary);
}
public async Task ConsumeAsync<T>(string queue, Func<T, Task> handler)
{
_channel.BasicConsume(
queue: queue,
autoAck: false,
consumer: new AsyncEventingBasicConsumer(_channel)
{
Received = async (ch, ea) =>
{
try
{
var message = _serializer.Deserialize<T>(ea.Body.ToArray());
await handler(message);
_channel.BasicAck(ea.DeliveryTag, false);
}
catch (Exception)
{
_channel.BasicNack(ea.DeliveryTag, false, true);
}
}
});
}
}public class FlexonServiceBus
{
private readonly ServiceBusClient _client;
private readonly IFlexonSerializer _serializer;
public async Task SendMessageAsync<T>(string queue, T message)
{
var sender = _client.CreateSender(queue);
var binary = _serializer.Serialize(message);
var sbMessage = new ServiceBusMessage(binary)
{
ContentType = "application/x-flexon",
TimeToLive = TimeSpan.FromDays(1)
};
await sender.SendMessageAsync(sbMessage);
}
public async Task ProcessMessagesAsync<T>(string queue,
Func<T, Task> handler)
{
var processor = _client.CreateProcessor(queue);
processor.ProcessMessageAsync += async args =>
{
var message = _serializer.Deserialize<T>(args.Message.Body.ToArray());
await handler(message);
};
processor.ProcessErrorAsync += args =>
{
// Handle error
return Task.CompletedTask;
};
await processor.StartProcessingAsync();
}
}public class FlexonBlobStorage
{
private readonly BlobServiceClient _blobService;
private readonly IFlexonSerializer _serializer;
public async Task UploadAsync<T>(string container, string blobName, T data)
{
var containerClient = _blobService.GetBlobContainerClient(container);
var blobClient = containerClient.GetBlobClient(blobName);
var binary = _serializer.Serialize(data);
using var stream = new MemoryStream(binary);
await blobClient.UploadAsync(stream, new BlobUploadOptions
{
HttpHeaders = new BlobHttpHeaders
{
ContentType = "application/x-flexon"
}
});
}
public async Task<T> DownloadAsync<T>(string container, string blobName)
{
var containerClient = _blobService.GetBlobContainerClient(container);
var blobClient = containerClient.GetBlobClient(blobName);
using var stream = new MemoryStream();
await blobClient.DownloadToAsync(stream);
return _serializer.Deserialize<T>(stream.ToArray());
}
}public class FlexonS3
{
private readonly IAmazonS3 _s3Client;
private readonly IFlexonSerializer _serializer;
public async Task UploadAsync<T>(string bucket, string key, T data)
{
var binary = _serializer.Serialize(data);
using var stream = new MemoryStream(binary);
await _s3Client.PutObjectAsync(new PutObjectRequest
{
BucketName = bucket,
Key = key,
InputStream = stream,
ContentType = "application/x-flexon"
});
}
public async Task<T> DownloadAsync<T>(string bucket, string key)
{
var response = await _s3Client.GetObjectAsync(bucket, key);
using var stream = response.ResponseStream;
using var ms = new MemoryStream();
await stream.CopyToAsync(ms);
return _serializer.Deserialize<T>(ms.ToArray());
}
}public class FlexonGrpcMarshaller<T>
{
private static readonly IFlexonSerializer _serializer = new FlexonSerializer();
public static Marshaller<T> Create() => new Marshaller<T>(
(T value, SerializationContext context) =>
{
var binary = _serializer.Serialize(value);
context.Complete(new ReadOnlyMemory<byte>(binary));
},
(DeserializationContext context) =>
{
var binary = context.PayloadAsNewBuffer();
return _serializer.Deserialize<T>(binary);
});
}
public static class FlexonGrpcServiceBase
{
public static Method<TRequest, TResponse> CreateMethod<TRequest, TResponse>(
MethodType methodType,
string serviceName,
string methodName)
{
return new Method<TRequest, TResponse>(
methodType,
serviceName,
methodName,
FlexonGrpcMarshaller<TRequest>.Create(),
FlexonGrpcMarshaller<TResponse>.Create());
}
}public class GameStateSync
{
private readonly IFlexonSerializer _serializer;
private readonly ConcurrentDictionary<string, GameState> _states;
public async Task BroadcastState(GameState state)
{
var options = new FlexonOptions
{
EnableCompression = true,
CompressionLevel = CompressionLevel.Fastest
};
var binary = _serializer.Serialize(state, options);
await BroadcastToClients(binary);
}
public async Task ProcessStateUpdate(byte[] binary)
{
var update = _serializer.Deserialize<StateUpdate>(binary);
await ApplyUpdate(update);
}
}public class IoTIntegration
{
private readonly IFlexonSerializer _serializer;
private readonly DeviceClient _deviceClient;
public async Task SendTelemetry(DeviceData data)
{
var options = new FlexonOptions
{
EnableCompression = true,
Schema = DeviceSchema.Instance
};
var binary = _serializer.Serialize(data, options);
var message = new Message(binary)
{
ContentType = "application/x-flexon",
ContentEncoding = "gzip"
};
await _deviceClient.SendEventAsync(message);
}
public async Task ProcessCommand(Message message)
{
var command = _serializer.Deserialize<DeviceCommand>(
message.GetBytes());
await ExecuteCommand(command);
}
}public class MLModelSerializer
{
private readonly IFlexonSerializer _serializer;
public async Task SaveModel(string path, MLModel model)
{
var options = new FlexonOptions
{
EnableCompression = true,
CompressionLevel = CompressionLevel.Optimal
};
var binary = _serializer.Serialize(model, options);
await File.WriteAllBytesAsync(path, binary);
}
public async Task<MLModel> LoadModel(string path)
{
var binary = await File.ReadAllBytesAsync(path);
return _serializer.Deserialize<MLModel>(binary);
}
}public class FeatureProcessor
{
private readonly IFlexonSerializer _serializer;
public async Task ProcessBatch(FeatureBatch batch)
{
var options = new FlexonOptions
{
EnableSimd = true,
UsePooledBuffers = true
};
var binary = _serializer.Serialize(batch.Features, options);
await ProcessFeatures(binary);
}
}-
Performance Optimization
- Use buffer pooling
- Enable compression when appropriate
- Configure proper batch sizes
- Monitor memory usage
-
Error Handling
- Implement proper retry logic
- Log serialization errors
- Handle version mismatches
- Validate data integrity
-
Security
- Validate input data
- Use encryption when needed
- Implement access control
- Monitor for attacks
-
Monitoring
- Track performance metrics
- Monitor error rates
- Log important operations
- Set up alerts