Browse Source

Message table in database changes the primary key to non auto-Increment. (#180)

master
Savorboard 6 years ago
parent
commit
c3b7d6b61f
11 changed files with 28 additions and 23 deletions
  1. +1
    -1
      src/DotNetCore.CAP.MySql/CapPublisher.cs
  2. +3
    -3
      src/DotNetCore.CAP.MySql/MySqlStorageConnection.cs
  3. +7
    -7
      src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs
  4. +2
    -4
      src/DotNetCore.CAP/IConsumerHandler.Default.cs
  5. +1
    -1
      src/DotNetCore.CAP/IStorageConnection.cs
  6. +1
    -0
      src/DotNetCore.CAP/Internal/ICallbackMessageSender.Default.cs
  7. +5
    -5
      src/DotNetCore.CAP/LoggerExtensions.cs
  8. +1
    -1
      src/DotNetCore.CAP/Models/CapPublishedMessage.cs
  9. +2
    -1
      src/DotNetCore.CAP/Models/CapReceivedMessage.cs
  10. +3
    -0
      test/DotNetCore.CAP.MySql.Test/MySqlStorageConnectionTest.cs
  11. +2
    -0
      test/DotNetCore.CAP.Test/Processor/StateChangerTest.cs

+ 1
- 1
src/DotNetCore.CAP.MySql/CapPublisher.cs View File

@@ -40,7 +40,7 @@ namespace DotNetCore.CAP.MySql
await PublishAsyncInternal(message);
}

protected override Task<int> ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
protected override Task ExecuteAsync(CapPublishedMessage message, ICapTransaction transaction,
CancellationToken cancel = default(CancellationToken))
{
var dbTrans = transaction.DbTransaction as IDbTransaction;


+ 3
- 3
src/DotNetCore.CAP.MySql/MySqlStorageConnection.cs View File

@@ -52,7 +52,7 @@ namespace DotNetCore.CAP.MySql
}
}

public async Task<int> StoreReceivedMessageAsync(CapReceivedMessage message)
public Task StoreReceivedMessageAsync(CapReceivedMessage message)
{
if (message == null)
{
@@ -61,11 +61,11 @@ namespace DotNetCore.CAP.MySql

var sql = $@"
INSERT INTO `{_prefix}.received`(`Name`,`Group`,`Content`,`Retries`,`Added`,`ExpiresAt`,`StatusName`)
VALUES(@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);SELECT LAST_INSERT_ID();";
VALUES(@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);";

using (var connection = new MySqlConnection(Options.ConnectionString))
{
return await connection.ExecuteScalarAsync<int>(sql, message);
return connection.ExecuteScalarAsync<int>(sql, message);
}
}



+ 7
- 7
src/DotNetCore.CAP/Abstractions/CapPublisherBase.cs View File

@@ -40,6 +40,7 @@ namespace DotNetCore.CAP.Abstractions
{
var message = new CapPublishedMessage
{
Id = SnowflakeId.Default().NextId(),
Name = name,
Content = Serialize(contentObj, callbackName),
StatusName = StatusName.Scheduled
@@ -53,6 +54,7 @@ namespace DotNetCore.CAP.Abstractions
{
var message = new CapPublishedMessage
{
Id = SnowflakeId.Default().NextId(),
Name = name,
Content = Serialize(contentObj, callbackName),
StatusName = StatusName.Scheduled
@@ -75,13 +77,11 @@ namespace DotNetCore.CAP.Abstractions
{
operationId = s_diagnosticListener.WritePublishMessageStoreBefore(message);

message.Id = await ExecuteAsync(message, CapTransaction);
await ExecuteAsync(message, CapTransaction);

if (message.Id > 0)
{
_capTransaction.AddToSent(message);
s_diagnosticListener.WritePublishMessageStoreAfter(operationId, message);
}
_capTransaction.AddToSent(message);

s_diagnosticListener.WritePublishMessageStoreAfter(operationId, message);

if (NotUseTransaction || CapTransaction.AutoCommit)
{
@@ -105,7 +105,7 @@ namespace DotNetCore.CAP.Abstractions

protected abstract object GetDbTransaction();

protected abstract Task<int> ExecuteAsync(CapPublishedMessage message,
protected abstract Task ExecuteAsync(CapPublishedMessage message,
ICapTransaction transaction,
CancellationToken cancel = default(CancellationToken));



+ 2
- 4
src/DotNetCore.CAP/IConsumerHandler.Default.cs View File

@@ -112,6 +112,7 @@ namespace DotNetCore.CAP

var receivedMessage = new CapReceivedMessage(messageContext)
{
Id = SnowflakeId.Default().NextId(),
StatusName = StatusName.Scheduled,
Content = messageBody
};
@@ -170,10 +171,7 @@ namespace DotNetCore.CAP

private void StoreMessage(CapReceivedMessage receivedMessage)
{
var id = _connection.StoreReceivedMessageAsync(receivedMessage)
.GetAwaiter().GetResult();

receivedMessage.Id = id;
_connection.StoreReceivedMessageAsync(receivedMessage).GetAwaiter().GetResult();
}

private (Guid, string) TracingBefore(string topic, string values)


+ 1
- 1
src/DotNetCore.CAP/IStorageConnection.cs View File

@@ -32,7 +32,7 @@ namespace DotNetCore.CAP
/// Stores the message.
/// </summary>
/// <param name="message">The message to store.</param>
Task<int> StoreReceivedMessageAsync(CapReceivedMessage message);
Task StoreReceivedMessageAsync(CapReceivedMessage message);

/// <summary>
/// Returns the message with the given id.


+ 1
- 0
src/DotNetCore.CAP/Internal/ICallbackMessageSender.Default.cs View File

@@ -54,6 +54,7 @@ namespace DotNetCore.CAP.Internal

var publishedMessage = new CapPublishedMessage
{
Id = SnowflakeId.Default().NextId(),
Name = topicName,
Content = content,
StatusName = StatusName.Scheduled


+ 5
- 5
src/DotNetCore.CAP/LoggerExtensions.cs View File

@@ -10,12 +10,12 @@ namespace DotNetCore.CAP
[SuppressMessage("ReSharper", "InconsistentNaming")]
internal static class LoggerExtensions
{
public static void ConsumerExecutedAfterThreshold(this ILogger logger, int messageId, int retries)
public static void ConsumerExecutedAfterThreshold(this ILogger logger, long messageId, int retries)
{
logger.LogWarning($"The Subscriber of the message({messageId}) still fails after {retries}th executions and we will stop retrying.");
}

public static void SenderAfterThreshold(this ILogger logger, int messageId, int retries)
public static void SenderAfterThreshold(this ILogger logger, long messageId, int retries)
{
logger.LogWarning($"The Publisher of the message({messageId}) still fails after {retries}th sends and we will stop retrying.");
}
@@ -25,12 +25,12 @@ namespace DotNetCore.CAP
logger.LogWarning(ex, "FailedThresholdCallback action raised an exception:" + ex.Message);
}

public static void ConsumerExecutionRetrying(this ILogger logger, int messageId, int retries)
public static void ConsumerExecutionRetrying(this ILogger logger, long messageId, int retries)
{
logger.LogWarning($"The {retries}th retrying consume a message failed. message id: {messageId}");
}

public static void SenderRetrying(this ILogger logger, int messageId, int retries)
public static void SenderRetrying(this ILogger logger, long messageId, int retries)
{
logger.LogWarning($"The {retries}th retrying send a message failed. message id: {messageId} ");
}
@@ -40,7 +40,7 @@ namespace DotNetCore.CAP
logger.LogDebug($"Message published. name: {name}, content:{content}.");
}

public static void MessagePublishException(this ILogger logger, int messageId, string reason, Exception ex)
public static void MessagePublishException(this ILogger logger, long messageId, string reason, Exception ex)
{
logger.LogError(ex, $"An exception occured while publishing a message, reason:{reason}. message id:{messageId}");
}


+ 1
- 1
src/DotNetCore.CAP/Models/CapPublishedMessage.cs View File

@@ -15,7 +15,7 @@ namespace DotNetCore.CAP.Models
Added = DateTime.Now;
}

public int Id { get; set; }
public long Id { get; set; }

public string Name { get; set; }



+ 2
- 1
src/DotNetCore.CAP/Models/CapReceivedMessage.cs View File

@@ -2,6 +2,7 @@
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using DotNetCore.CAP.Infrastructure;

namespace DotNetCore.CAP.Models
{
@@ -22,7 +23,7 @@ namespace DotNetCore.CAP.Models
Content = message.Content;
}

public int Id { get; set; }
public long Id { get; set; }

public string Group { get; set; }



+ 3
- 0
test/DotNetCore.CAP.MySql.Test/MySqlStorageConnectionTest.cs View File

@@ -25,6 +25,7 @@ namespace DotNetCore.CAP.MySql.Test
var sql = "INSERT INTO `cap.published`(`Name`,`Content`,`Retries`,`Added`,`ExpiresAt`,`StatusName`) VALUES(@Name,@Content,@Retries,@Added,@ExpiresAt,@StatusName);SELECT @@IDENTITY;";
var publishMessage = new CapPublishedMessage
{
Id = SnowflakeId.Default().NextId(),
Name = "MySqlStorageConnectionTest",
Content = "",
StatusName = StatusName.Scheduled
@@ -45,6 +46,7 @@ namespace DotNetCore.CAP.MySql.Test
{
var receivedMessage = new CapReceivedMessage
{
Id = SnowflakeId.Default().NextId(),
Name = "MySqlStorageConnectionTest",
Content = "",
Group = "mygroup",
@@ -71,6 +73,7 @@ namespace DotNetCore.CAP.MySql.Test
VALUES(@Name,@Group,@Content,@Retries,@Added,@ExpiresAt,@StatusName);SELECT @@IDENTITY;";
var receivedMessage = new CapReceivedMessage
{
Id = SnowflakeId.Default().NextId(),
Name = "MySqlStorageConnectionTest",
Content = "",
Group = "mygroup",


+ 2
- 0
test/DotNetCore.CAP.Test/Processor/StateChangerTest.cs View File

@@ -16,6 +16,7 @@ namespace DotNetCore.CAP.Test
var fixture = Create();
var message = new CapPublishedMessage
{
Id = SnowflakeId.Default().NextId(),
StatusName = StatusName.Scheduled
};
var state = Mock.Of<IState>(s => s.Name == "s" && s.ExpiresAfter == null);
@@ -39,6 +40,7 @@ namespace DotNetCore.CAP.Test
var fixture = Create();
var message = new CapPublishedMessage
{
Id = SnowflakeId.Default().NextId(),
StatusName = StatusName.Scheduled
};
var state = Mock.Of<IState>(s => s.Name == "s" && s.ExpiresAfter == TimeSpan.FromHours(1));


Loading…
Cancel
Save