测试可以发送，接收长度小于100的字符串消息，长消息，会崩溃。
PublicDependencyModuleNames.AddRange(new string[] { "Core", "CoreUObject", "Engine", "InputCore", "EnhancedInput", "WebSockets", "Json", "JsonUtilities", "MQTTCore", "AudioMixer"});MyObject类代码：
MyObject.h
// Fill out your copyright notice in the Description page of Project Settings. #pragma once #include "CoreMinimal.h" #include "UObject/NoExportTypes.h" #include "IMQTTClient.h" #include "Json.h" #include "JsonUtilities.h" #include "Misc/Base64.h" #include "Misc/FileHelper.h" #include "MyObject.generated.h" UCLASS(BlueprintType, Blueprintable) class METAHUMANCHARACTERHEIXI_API UMyObject : public UObject { GENERATED_BODY() public: //UPROPERTY() TSharedPtr<IMQTTClient, ESPMode::ThreadSafe> MQTTClient; UFUNCTION(BlueprintCallable, Category = "Demo") void HelloWorld(); void SaveWav(const FString& FilePath, const TArray<uint8>& AudioBytes, int32 SampleRate, int32 NumChannels); void AppendAudioChunk(const TArray<uint8>& Chunk); TArray<uint8> ReceivedAudioBuffer; protected: virtual void BeginDestroy() override; };MyObject.cpp
// Fill out your copyright notice in the Description page of Project Settings. #include "MyObject.h" #include "MQTTSubsystem.h" #include "IMQTTCoreModule.h" #include "MQTTClientSettings.h" #include "Async/Async.h" #include "MQTTShared.h" #include "Engine/Engine.h" //TSharedPtr<IMQTTClient, ESPMode::ThreadSafe> UMyObject::MQTTClient = nullptr; void UMyObject::SaveWav(const FString& FilePath, const TArray<uint8>& AudioBytes, int32 SampleRate, int32 NumChannels) { if (AudioBytes.Num() == 0) return; const int32 BitsPerSample = 16; const int32 BlockAlign = NumChannels * BitsPerSample / 8; const int32 ByteRate = SampleRate * BlockAlign; const int32 DataSize = AudioBytes.Num(); const int32 ChunkSize = 36 + DataSize; TArray<uint8> Wav; Wav.Reserve(44 + DataSize); // 头44字节 + PCM数据 auto AppendInt32 = [&Wav](int32 V) { Wav.Append(reinterpret_cast<uint8*>(&V), sizeof(int32)); }; auto AppendInt16 = [&Wav](int16 V) { Wav.Append(reinterpret_cast<uint8*>(&V), sizeof(int16)); }; // --- WAV Header --- Wav.Append(reinterpret_cast<const uint8*>("RIFF"), 4); AppendInt32(ChunkSize); Wav.Append(reinterpret_cast<const uint8*>("WAVE"), 4); Wav.Append(reinterpret_cast<const uint8*>("fmt "), 4); AppendInt32(16); // fmt chunk size AppendInt16(1); // PCM format AppendInt16(NumChannels); AppendInt32(SampleRate); AppendInt32(ByteRate); AppendInt16(BlockAlign); AppendInt16(BitsPerSample); Wav.Append(reinterpret_cast<const uint8*>("data"), 4); AppendInt32(DataSize); // --- PCM Data --- Wav.Append(AudioBytes); FFileHelper::SaveArrayToFile(Wav, *FilePath); } void UMyObject::AppendAudioChunk(const TArray<uint8>& Chunk) { ReceivedAudioBuffer.Append(Chunk); } void UMyObject::HelloWorld() { UE_LOG(LogTemp, Warning, TEXT("Hello World from UMyObject!")); if (GEngine) { GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Green, TEXT("Hello World from UMyObject!")); } IMQTTCoreModule& MQTTModule = FModuleManager::LoadModuleChecked<IMQTTCoreModule>("MQTTCore"); FMQTTURL URL; URL.Host = TEXT("127.0.0.1"); URL.Port = 1883; MQTTClient = MQTTModule.GetOrCreateClient(URL); if (!MQTTClient.IsValid()) { UE_LOG(LogTemp, Error, TEXT("MQTTClient 无效！")); return; } TSharedPtr<IMQTTClient> LocalClient = MQTTClient; MQTTClient->OnMessage().AddLambda([](const FMQTTClientMessage& Msg) { // 拷贝数据 FString TopicCopy = Msg.Topic; TArray<uint8> PayloadCopy = Msg.Payload; // 切回 GameThread，显式 Move 捕获 AsyncTask(ENamedThreads::GameThread, [TopicCopy = MoveTemp(TopicCopy), PayloadCopy = MoveTemp(PayloadCopy)]() { if (TopicCopy == "ue/audio") { FString FilePath = FPaths::ProjectSavedDir() / TEXT("received.wav"); if (PayloadCopy.Num() > 0) { UE_LOG(LogTemp, Error, TEXT("[MQTT] start Saved audio ")); const int32 SampleRate = 44100; const int32 NumChannels = 1; const int32 BitsPerSample = 16; const int32 BlockAlign = NumChannels * BitsPerSample / 8; const int32 ByteRate = SampleRate * BlockAlign; const int32 DataSize = PayloadCopy.Num(); const int32 ChunkSize = 36 + DataSize; TArray<uint8> Wav; Wav.Reserve(44 + DataSize); auto AppendInt32 = [&Wav](int32 V) { Wav.Append(reinterpret_cast<uint8*>(&V), sizeof(int32)); }; auto AppendInt16 = [&Wav](int16 V) { Wav.Append(reinterpret_cast<uint8*>(&V), sizeof(int16)); }; // WAV header Wav.Append(reinterpret_cast<const uint8*>("RIFF"), 4); AppendInt32(ChunkSize); Wav.Append(reinterpret_cast<const uint8*>("WAVE"), 4); Wav.Append(reinterpret_cast<const uint8*>("fmt "), 4); AppendInt32(16); AppendInt16(1); AppendInt16(NumChannels); AppendInt32(SampleRate); AppendInt32(ByteRate); AppendInt16(BlockAlign); AppendInt16(BitsPerSample); Wav.Append(reinterpret_cast<const uint8*>("data"), 4); AppendInt32(DataSize); // PCM 数据 Wav.Append(PayloadCopy); FFileHelper::SaveArrayToFile(Wav, *FilePath); UE_LOG(LogTemp, Error, TEXT("[MQTT] Saved audio, size=%d bytes"), DataSize); } } else if (TopicCopy == "ue/messages") { //FString PayloadStr(UTF8_TO_TCHAR(PayloadCopy.GetData())); const uint8* DataPtr = PayloadCopy.GetData(); int32 DataSize = PayloadCopy.Num(); FString PayloadStr(DataSize, UTF8_TO_TCHAR((const char*)DataPtr)); TSharedPtr<FJsonObject> JsonObject; TSharedRef<TJsonReader<>> Reader = TJsonReaderFactory<>::Create(PayloadStr); if (FJsonSerializer::Deserialize(Reader, JsonObject) && JsonObject.IsValid()) { FString Type = JsonObject->GetStringField("type"); if (Type == "command") { FString CommandData = JsonObject->GetStringField("data"); UE_LOG(LogTemp, Log, TEXT("[MQTT] Command: %s"), *CommandData); if (GEngine) { GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Yellow, FString::Printf(TEXT("MQTT Command: %s"), *CommandData)); } } else { UE_LOG(LogTemp, Error, TEXT("[MQTT] DataSize: %d"), DataSize); } } } }); }); MQTTClient->OnConnect().AddLambda([LocalClient](EMQTTConnectReturnCode ReturnCode) { // 后台线程回调 AsyncTask(ENamedThreads::GameThread, [ReturnCode, LocalClient]() { if (ReturnCode == EMQTTConnectReturnCode::Accepted) { UE_LOG(LogTemp, Warning, TEXT("MQTT Connected!")); if (GEngine) { GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Green, TEXT("conn ok")); } //TArray<TPair<FString, EMQTTQualityOfService>> TopicsToSubscribe; //TopicsToSubscribe.Add(MakeTuple(FString("ue/messages"), EMQTTQualityOfService::Once)); //TopicsToSubscribe.Add(MakeTuple(FString("ue/audio"), EMQTTQualityOfService::Once)); //// 订阅 topic //if (UMyObject::MQTTClient.IsValid()) //{ // UMyObject::MQTTClient->Subscribe(TopicsToSubscribe); // if (GEngine) // { // GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Green, TEXT("MQTT TopicsToSubscribe ok")); // } //} TArray<TPair<FString, EMQTTQualityOfService>> TopicsToSubscribe; TopicsToSubscribe.Add(MakeTuple(FString("ue/messages"), EMQTTQualityOfService::Once)); TopicsToSubscribe.Add(MakeTuple(FString("ue/audio"), EMQTTQualityOfService::Once)); LocalClient->Subscribe(TopicsToSubscribe); UE_LOG(LogTemp, Error, TEXT("MQTT TopicsToSubscribe ok")); if (GEngine) { GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Green, TEXT("MQTT TopicsToSubscribe ok")); } /* if (LocalClient.IsValid()) { FTimerHandle TimerHandle; GWorld->GetTimerManager().SetTimer(TimerHandle, [LocalClient]() { if (LocalClient.IsValid()) { TArray<TPair<FString, EMQTTQualityOfService>> TopicsToSubscribe; TopicsToSubscribe.Add(MakeTuple(FString("ue/messages"), EMQTTQualityOfService::Once)); TopicsToSubscribe.Add(MakeTuple(FString("ue/audio"), EMQTTQualityOfService::Once)); LocalClient->Subscribe(TopicsToSubscribe); UE_LOG(LogTemp, Error, TEXT("MQTT TopicsToSubscribe ok")); if (GEngine) { GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Green, TEXT("MQTT TopicsToSubscribe ok")); } } }, 0.1f, false); }*/ } else { UE_LOG(LogTemp, Error, TEXT("MQTT con failed，ReturnCode=%d"), static_cast<int32>(ReturnCode)); } }); }); /*MQTTClient->OnConnect().AddLambda([LocalClient](EMQTTConnectReturnCode ReturnCode) { if (ReturnCode == EMQTTConnectReturnCode::Accepted) { UE_LOG(LogTemp, Error, TEXT("MQTT conn ok")); TArray<TPair<FString, EMQTTQualityOfService>> TopicsToSubscribe; TopicsToSubscribe.Add(MakeTuple(FString("ue/messages"), EMQTTQualityOfService::Once)); TopicsToSubscribe.Add(MakeTuple(FString("ue/audio"), EMQTTQualityOfService::Once)); LocalClient->Subscribe(TopicsToSubscribe); UE_LOG(LogTemp, Error, TEXT("MQTT TopicsToSubscribe ok")); if (GEngine) { GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Green, TEXT("MQTT TopicsToSubscribe ok")); } } else { UE_LOG(LogTemp, Warning, TEXT("MQTT con failed，ReturnCode=%d"), static_cast<int32>(ReturnCode)); } });*/ /*FTimerHandle TimerHandle; GetWorld()->GetTimerManager().SetTimer(TimerHandle, [this]() { if (MQTTClient.IsValid()) { MQTTClient->Connect(); } }, 0.1f, false);*/ MQTTClient->Connect(); if (GEngine) { GEngine->AddOnScreenDebugMessage(-1, 5.f, FColor::Green, TEXT("IMQTTCoreModule 127.0.0.1----------!")); } } void UMyObject::BeginDestroy() { // 确保后台线程不再访问 MQTTClient if (MQTTClient.IsValid()) { // 断开连接 MQTTClient->Disconnect(); UE_LOG(LogTemp, Error, TEXT("MQTT Disconnect")); MQTTClient.Reset(); } Super::BeginDestroy(); }
